opendaylight/md-sal/samples/clustering-test-app/provider/src/main/java/org/opendaylight/controller/clustering/it/provider/impl/PublishNotificationsTask.java

   1 /*
   2  * Copyright (c) 2017 Cisco Systems, Inc. and others.  All rights reserved.
   3  *
   4  * This program and the accompanying materials are made available under the
   5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   6  * and is available at http://www.eclipse.org/legal/epl-v10.html
   7  */
   8
   9 package org.opendaylight.controller.clustering.it.provider.impl;
  10
  11 import com.google.common.base.Preconditions;
  12 import com.google.common.util.concurrent.SettableFuture;
  13 import java.util.concurrent.Executors;
  14 import java.util.concurrent.ScheduledExecutorService;
  15 import java.util.concurrent.ScheduledFuture;
  16 import java.util.concurrent.TimeUnit;
  17 import org.opendaylight.controller.md.sal.binding.api.NotificationPublishService;
  18 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.target.rev170215.IdSequence;
  19 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.target.rev170215.IdSequenceBuilder;
  20 import org.opendaylight.yangtools.yang.common.RpcError;
  21 import org.opendaylight.yangtools.yang.common.RpcResult;
  22 import org.opendaylight.yangtools.yang.common.RpcResultBuilder;
  23 import org.slf4j.Logger;
  24 import org.slf4j.LoggerFactory;
  25
  26 public class PublishNotificationsTask implements Runnable {
  27
  28     private static final Logger LOG = LoggerFactory.getLogger(PublishNotificationsTask.class);
  29     private static final int SECOND_AS_NANO = 1000000000;
  30
  31     private final NotificationPublishService notificationPublishService;
  32     private final String notificationId;
  33     private final long timeToTake;
  34     private final long delay;
  35
  36     private final ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor();
  37
  38     private long sequenceNumber = 1;
  39     private long startTime;
  40     private SettableFuture<RpcResult<Void>> completionFuture;
  41     private ScheduledFuture<?> scheduledFuture;
  42
  43     public PublishNotificationsTask(final NotificationPublishService notificationPublishService,
  44                                     final String notificationId, final long secondsToTake, final long maxPerSecond) {
  45         Preconditions.checkNotNull(notificationPublishService);
  46         Preconditions.checkNotNull(notificationId);
  47         Preconditions.checkArgument(secondsToTake > 0);
  48         Preconditions.checkArgument(maxPerSecond > 0);
  49
  50         this.notificationPublishService = notificationPublishService;
  51         this.notificationId = notificationId;
  52         this.timeToTake = secondsToTake * SECOND_AS_NANO;
  53         this.delay = SECOND_AS_NANO / maxPerSecond;
  54
  55         LOG.debug("Delay : {}", delay);
  56     }
  57
  58     @Override
  59     public void run() {
  60         final long current = System.nanoTime();
  61
  62         final IdSequence notification =
  63                 new IdSequenceBuilder().setId(notificationId).setSequenceNumber(sequenceNumber).build();
  64         sequenceNumber++;
  65
  66         try {
  67             LOG.debug("Publishing notification: {}", notification);
  68             notificationPublishService.putNotification(notification);
  69         } catch (final InterruptedException e) {
  70             LOG.warn("Unexpected exception while publishing notification, : {}", notification, e);
  71             completionFuture.set(RpcResultBuilder.<Void>failed()
  72                     .withError(RpcError.ErrorType.APPLICATION, "Unexpected-exception", e).build());
  73         }
  74
  75         LOG.debug("current {}, starttime: {}, timetotake: {}, current-start = {}",
  76                 current, startTime, timeToTake, current - startTime);
  77
  78         if ((current - startTime) > timeToTake) {
  79             completionFuture.set(RpcResultBuilder.<Void>success().build());
  80             LOG.debug("Sequence number: {}", sequenceNumber);
  81             scheduledFuture.cancel(false);
  82             executor.shutdown();
  83         }
  84     }
  85
  86     public void start(final SettableFuture<RpcResult<Void>> settableFuture) {
  87         startTime = System.nanoTime();
  88         completionFuture = settableFuture;
  89         scheduledFuture = executor.scheduleAtFixedRate(this, 0, delay, TimeUnit.NANOSECONDS);
  90     }
  91 }