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 java.util.concurrent.Executors;
  13 import java.util.concurrent.ScheduledExecutorService;
  14 import java.util.concurrent.ScheduledFuture;
  15 import java.util.concurrent.TimeUnit;
  16 import org.opendaylight.controller.md.sal.binding.api.NotificationPublishService;
  17 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.target.rev170215.IdSequence;
  18 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.target.rev170215.IdSequenceBuilder;
  19 import org.slf4j.Logger;
  20 import org.slf4j.LoggerFactory;
  21
  22 public class PublishNotificationsTask implements Runnable {
  23
  24     private static final Logger LOG = LoggerFactory.getLogger(PublishNotificationsTask.class);
  25     private static final int SECOND_AS_NANO = 1000000000;
  26
  27     private final NotificationPublishService notificationPublishService;
  28     private final String notificationId;
  29     private final long timeToTake;
  30     private final long delay;
  31
  32     private final ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor();
  33
  34     private long sequenceNumber = 1;
  35     private long startTime;
  36     private ScheduledFuture<?> scheduledFuture;
  37
  38     private Exception lastError = null;
  39
  40     public PublishNotificationsTask(final NotificationPublishService notificationPublishService,
  41                                     final String notificationId, final long secondsToTake, final long maxPerSecond) {
  42         Preconditions.checkNotNull(notificationPublishService);
  43         Preconditions.checkNotNull(notificationId);
  44         Preconditions.checkArgument(secondsToTake > 0);
  45         Preconditions.checkArgument(maxPerSecond > 0);
  46
  47         this.notificationPublishService = notificationPublishService;
  48         this.notificationId = notificationId;
  49         this.timeToTake = secondsToTake * SECOND_AS_NANO;
  50         this.delay = SECOND_AS_NANO / maxPerSecond;
  51
  52         LOG.debug("Delay : {}", delay);
  53     }
  54
  55     @Override
  56     public void run() {
  57         final long current = System.nanoTime();
  58
  59         final IdSequence notification =
  60                 new IdSequenceBuilder().setId(notificationId).setSequenceNumber(sequenceNumber).build();
  61         sequenceNumber++;
  62
  63         try {
  64             LOG.debug("Publishing notification: {}", notification);
  65             notificationPublishService.putNotification(notification);
  66         } catch (final InterruptedException e) {
  67             LOG.warn("Unexpected exception while publishing notification, : {}", notification, e);
  68             lastError = e;
  69
  70             //stop on error
  71             scheduledFuture.cancel(false);
  72             executor.shutdown();
  73             return;
  74         }
  75
  76         LOG.debug("current {}, starttime: {}, timetotake: {}, current-start = {}",
  77                 current, startTime, timeToTake, current - startTime);
  78
  79         if ((current - startTime) > timeToTake) {
  80             LOG.debug("Sequence number: {}", sequenceNumber);
  81             scheduledFuture.cancel(false);
  82             executor.shutdown();
  83         }
  84     }
  85
  86     public void start() {
  87         startTime = System.nanoTime();
  88         scheduledFuture = executor.scheduleAtFixedRate(this, 0, delay, TimeUnit.NANOSECONDS);
  89     }
  90
  91     public boolean isFinished() {
  92         return scheduledFuture.isCancelled();
  93     }
  94
  95     public long getCurrentNotif() {
  96         return sequenceNumber;
  97     }
  98
  99     public Exception getLastError() {
 100         return lastError;
 101     }
 102 }