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

   1 /*
   2  * Copyright (c) 2017 Pantheon Technologies, s.r.o. 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 com.google.common.base.Stopwatch;
  11 import com.google.common.util.concurrent.ListenableFuture;
  12 import com.google.common.util.concurrent.SettableFuture;
  13 import java.util.Collection;
  14 import java.util.Iterator;
  15 import java.util.concurrent.Executors;
  16 import java.util.concurrent.ScheduledExecutorService;
  17 import java.util.concurrent.ScheduledFuture;
  18 import java.util.concurrent.TimeUnit;
  19 import javax.annotation.Nonnull;
  20 import javax.annotation.concurrent.GuardedBy;
  21 import org.opendaylight.yangtools.concepts.ListenerRegistration;
  22 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeCandidate;
  23 import org.slf4j.Logger;
  24 import org.slf4j.LoggerFactory;
  25
  26 public abstract class AbstractDataListener {
  27     private static final Logger LOG = LoggerFactory.getLogger(AbstractDataListener.class);
  28     private static final long MAX_ELAPSED_NANOS = TimeUnit.SECONDS.toNanos(4);
  29
  30     @GuardedBy("ticksSinceLast")
  31     private final Stopwatch ticksSinceLast = Stopwatch.createUnstarted();
  32
  33     private DataListenerState state = DataListenerState.initial();
  34     private ScheduledFuture<?> scheduledFuture;
  35
  36     AbstractDataListener() {
  37         // Do not allow instantiation from outside of the class
  38     }
  39
  40     public final ListenableFuture<DataListenerState> tryFinishProcessing(final ListenerRegistration<?> ddtlReg) {
  41         final ScheduledExecutorService executorService = Executors.newSingleThreadScheduledExecutor();
  42         final SettableFuture<DataListenerState> future = SettableFuture.create();
  43
  44         scheduledFuture = executorService.scheduleAtFixedRate(() -> {
  45             final long elapsed;
  46             synchronized (ticksSinceLast) {
  47                 elapsed = ticksSinceLast.elapsed(TimeUnit.NANOSECONDS);
  48             }
  49
  50             if (elapsed > MAX_ELAPSED_NANOS) {
  51                 ddtlReg.close();
  52                 future.set(state);
  53                 scheduledFuture.cancel(false);
  54                 executorService.shutdown();
  55             }
  56         }, 0, 1, TimeUnit.SECONDS);
  57         return future;
  58     }
  59
  60     final void onReceivedChanges(@Nonnull final Collection<DataTreeCandidate> changes) {
  61         // do not log the change into debug, only use trace since it will lead to OOM on default heap settings
  62         LOG.debug("Received {} data tree changed events", changes.size());
  63         changes.forEach(change -> {
  64             LOG.trace("Processing change {}", change);
  65             state = state.append(change);
  66         });
  67
  68         synchronized (this) {
  69             ticksSinceLast.reset().start();
  70         }
  71     }
  72
  73     final void onReceivedError(final Collection<? extends Throwable> errors) {
  74         final Iterator<? extends Throwable> it = errors.iterator();
  75         final Throwable first = it.next();
  76         it.forEachRemaining(first::addSuppressed);
  77
  78         LOG.error("Listener failed", first);
  79
  80         // FIXME: mark the failure
  81     }
  82 }