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 package org.opendaylight.controller.clustering.it.provider.impl;
   9
  10 import static com.google.common.base.Preconditions.checkArgument;
  11 import static java.util.Objects.requireNonNull;
  12
  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.mdsal.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.slf4j.Logger;
  21 import org.slf4j.LoggerFactory;
  22
  23 public final class PublishNotificationsTask implements Runnable {
  24     private static final Logger LOG = LoggerFactory.getLogger(PublishNotificationsTask.class);
  25     private static final int SECOND_AS_NANO = 1_000_000_000;
  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
  43         this.notificationPublishService = requireNonNull(notificationPublishService);
  44         this.notificationId = requireNonNull(notificationId);
  45         checkArgument(secondsToTake > 0);
  46         timeToTake = secondsToTake * SECOND_AS_NANO;
  47         checkArgument(maxPerSecond > 0);
  48         delay = SECOND_AS_NANO / maxPerSecond;
  49
  50         LOG.debug("Delay : {}", delay);
  51     }
  52
  53     @Override
  54     public void run() {
  55         final long current = System.nanoTime();
  56
  57         final IdSequence notification =
  58                 new IdSequenceBuilder().setId(notificationId).setSequenceNumber(sequenceNumber).build();
  59         sequenceNumber++;
  60
  61         try {
  62             LOG.debug("Publishing notification: {}", notification);
  63             notificationPublishService.putNotification(notification);
  64         } catch (final InterruptedException e) {
  65             LOG.warn("Unexpected exception while publishing notification, : {}", notification, e);
  66             lastError = e;
  67
  68             //stop on error
  69             scheduledFuture.cancel(false);
  70             executor.shutdown();
  71             return;
  72         }
  73
  74         LOG.debug("current {}, starttime: {}, timetotake: {}, current-start = {}",
  75                 current, startTime, timeToTake, current - startTime);
  76
  77         if (current - startTime > timeToTake) {
  78             LOG.debug("Sequence number: {}", sequenceNumber);
  79             scheduledFuture.cancel(false);
  80             executor.shutdown();
  81         }
  82     }
  83
  84     public void start() {
  85         startTime = System.nanoTime();
  86         scheduledFuture = executor.scheduleAtFixedRate(this, 0, delay, TimeUnit.NANOSECONDS);
  87     }
  88
  89     public boolean isFinished() {
  90         return scheduledFuture.isCancelled();
  91     }
  92
  93     public long getCurrentNotif() {
  94         return sequenceNumber;
  95     }
  96
  97     public Exception getLastError() {
  98         return lastError;
  99     }
 100 }