package org.opendaylight.openflowplugin.impl.statistics;
import com.google.common.annotations.VisibleForTesting;
-import com.google.common.base.Function;
import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableList;
-import com.google.common.collect.Iterators;
-import com.google.common.util.concurrent.AsyncFunction;
import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
-import io.netty.util.Timeout;
+import com.google.common.util.concurrent.ListeningExecutorService;
+import com.google.common.util.concurrent.MoreExecutors;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashSet;
-import java.util.Iterator;
import java.util.List;
import java.util.Objects;
+import java.util.Optional;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
import javax.annotation.Nonnull;
import javax.annotation.Nullable;
-import javax.annotation.concurrent.GuardedBy;
import org.opendaylight.mdsal.common.api.TransactionChainClosedException;
import org.opendaylight.mdsal.singleton.common.api.ServiceGroupIdentifier;
import org.opendaylight.openflowplugin.api.ConnectionException;
import org.opendaylight.openflowplugin.api.openflow.device.DeviceInfo;
import org.opendaylight.openflowplugin.api.openflow.device.DeviceState;
import org.opendaylight.openflowplugin.api.openflow.device.RequestContext;
-import org.opendaylight.openflowplugin.api.openflow.device.handlers.ClusterInitializationPhaseHandler;
import org.opendaylight.openflowplugin.api.openflow.lifecycle.ContextChainMastershipState;
-import org.opendaylight.openflowplugin.api.openflow.lifecycle.MastershipChangeListener;
-import org.opendaylight.openflowplugin.api.openflow.rpc.listener.ItemLifecycleListener;
+import org.opendaylight.openflowplugin.api.openflow.lifecycle.ContextChainMastershipWatcher;
import org.opendaylight.openflowplugin.api.openflow.statistics.StatisticsContext;
-import org.opendaylight.openflowplugin.api.openflow.statistics.StatisticsManager;
import org.opendaylight.openflowplugin.impl.datastore.MultipartWriterProvider;
import org.opendaylight.openflowplugin.impl.rpc.AbstractRequestContext;
-import org.opendaylight.openflowplugin.impl.rpc.listener.ItemLifecycleListenerImpl;
import org.opendaylight.openflowplugin.impl.services.util.RequestContextUtil;
import org.opendaylight.openflowplugin.impl.statistics.services.dedicated.StatisticsGatheringOnTheFlyService;
import org.opendaylight.openflowplugin.impl.statistics.services.dedicated.StatisticsGatheringService;
private static final Logger LOG = LoggerFactory.getLogger(StatisticsContextImpl.class);
private static final String CONNECTION_CLOSED = "Connection closed.";
- private final ItemLifecycleListener itemLifeCycleListener;
private final Collection<RequestContext<?>> requestContexts = new HashSet<>();
private final DeviceContext deviceContext;
private final DeviceState devState;
+ private final ListeningExecutorService executorService;
private final boolean isStatisticsPollingOn;
- private final Object collectionStatTypeLock = new Object();
private final ConvertorExecutor convertorExecutor;
private final MultipartWriterProvider statisticsWriterProvider;
- @GuardedBy("collectionStatTypeLock")
+ private final DeviceInfo deviceInfo;
+ private final TimeCounter timeCounter = new TimeCounter();
+ private final long statisticsPollingInterval;
+ private final long maximumPollingDelay;
+ private final boolean isUsingReconciliationFramework;
+ private final AtomicBoolean schedulingEnabled = new AtomicBoolean(true);
+ private final AtomicReference<ListenableFuture<Boolean>> lastDataGatheringRef = new AtomicReference<>();
+ private final AtomicReference<StatisticsPollingService> statisticsPollingServiceRef = new AtomicReference<>();
private List<MultipartType> collectingStatType;
-
private StatisticsGatheringService<T> statisticsGatheringService;
private StatisticsGatheringOnTheFlyService<T> statisticsGatheringOnTheFlyService;
- private Timeout pollTimeout;
- private final DeviceInfo deviceInfo;
- private final StatisticsManager myManager;
-
- private volatile boolean schedulingEnabled;
- private volatile CONTEXT_STATE state;
- private ClusterInitializationPhaseHandler clusterInitializationPhaseHandler;
- private ClusterInitializationPhaseHandler initialSubmitHandler;
-
- private volatile ListenableFuture<Boolean> lastDataGathering;
+ private ContextChainMastershipWatcher contextChainMastershipWatcher;
- StatisticsContextImpl(final boolean isStatisticsPollingOn,
- @Nonnull final DeviceContext deviceContext,
+ StatisticsContextImpl(@Nonnull final DeviceContext deviceContext,
@Nonnull final ConvertorExecutor convertorExecutor,
- @Nonnull final StatisticsManager myManager,
- @Nonnull final MultipartWriterProvider statisticsWriterProvider) {
+ @Nonnull final MultipartWriterProvider statisticsWriterProvider,
+ @Nonnull final ListeningExecutorService executorService, boolean isStatisticsPollingOn,
+ boolean isUsingReconciliationFramework, long statisticsPollingInterval,
+ long maximumPollingDelay) {
this.deviceContext = deviceContext;
this.devState = Preconditions.checkNotNull(deviceContext.getDeviceState());
+ this.executorService = executorService;
this.isStatisticsPollingOn = isStatisticsPollingOn;
this.convertorExecutor = convertorExecutor;
- statisticsGatheringService = new StatisticsGatheringService<>(this, deviceContext);
- statisticsGatheringOnTheFlyService = new StatisticsGatheringOnTheFlyService<>(this,
- deviceContext, convertorExecutor, statisticsWriterProvider);
- itemLifeCycleListener = new ItemLifecycleListenerImpl(deviceContext);
- statListForCollectingInitialization();
- this.state = CONTEXT_STATE.INITIALIZATION;
this.deviceInfo = deviceContext.getDeviceInfo();
- this.myManager = myManager;
- this.lastDataGathering = null;
+ this.statisticsPollingInterval = statisticsPollingInterval;
+ this.maximumPollingDelay = maximumPollingDelay;
this.statisticsWriterProvider = statisticsWriterProvider;
+ this.isUsingReconciliationFramework = isUsingReconciliationFramework;
+
+ statisticsGatheringService = new StatisticsGatheringService<>(this, deviceContext);
+ statisticsGatheringOnTheFlyService = new StatisticsGatheringOnTheFlyService<>(this, deviceContext,
+ convertorExecutor,
+ statisticsWriterProvider);
}
@Override
- public void statListForCollectingInitialization() {
- synchronized (collectionStatTypeLock) {
- final List<MultipartType> statListForCollecting = new ArrayList<>();
- if (devState.isTableStatisticsAvailable()) {
- statListForCollecting.add(MultipartType.OFPMPTABLE);
- }
- if (devState.isFlowStatisticsAvailable()) {
- statListForCollecting.add(MultipartType.OFPMPFLOW);
- }
- if (devState.isGroupAvailable()) {
- statListForCollecting.add(MultipartType.OFPMPGROUPDESC);
- statListForCollecting.add(MultipartType.OFPMPGROUP);
- }
- if (devState.isMetersAvailable()) {
- statListForCollecting.add(MultipartType.OFPMPMETERCONFIG);
- statListForCollecting.add(MultipartType.OFPMPMETER);
- }
- if (devState.isPortStatisticsAvailable()) {
- statListForCollecting.add(MultipartType.OFPMPPORTSTATS);
- }
- if (devState.isQueueStatisticsAvailable()) {
- statListForCollecting.add(MultipartType.OFPMPQUEUE);
- }
- collectingStatType = ImmutableList.copyOf(statListForCollecting);
- }
+ public DeviceInfo getDeviceInfo() {
+ return this.deviceInfo;
}
+ @Nonnull
@Override
- public ListenableFuture<Boolean> gatherDynamicData() {
- if (!isStatisticsPollingOn) {
- LOG.debug("Statistics for device {} is not enabled.", getDeviceInfo().getNodeId().getValue());
- return Futures.immediateFuture(Boolean.TRUE);
- }
-
- if (Objects.isNull(lastDataGathering)
- || lastDataGathering.isCancelled()
- || lastDataGathering.isDone()) {
- lastDataGathering = Futures.immediateFuture(Boolean.TRUE);
- }
-
- synchronized (collectionStatTypeLock) {
- // write start timestamp to state snapshot container
- StatisticsGatheringUtils.markDeviceStateSnapshotStart(deviceContext);
-
- lastDataGathering = collectingStatType.stream().reduce(
- lastDataGathering,
- this::statChainFuture,
- (a, b) -> Futures.transform(a, (AsyncFunction<Boolean, Boolean>) result -> b));
-
- // write end timestamp to state snapshot container
- Futures.addCallback(lastDataGathering, new FutureCallback<Boolean>() {
- @Override
- public void onSuccess(final Boolean result) {
- StatisticsGatheringUtils.markDeviceStateSnapshotEnd(deviceContext, result);
- }
-
- @Override
- public void onFailure(final Throwable t) {
- if (!(t instanceof TransactionChainClosedException)) {
- StatisticsGatheringUtils.markDeviceStateSnapshotEnd(deviceContext, false);
- }
- }
- });
- }
-
- return lastDataGathering;
+ public ServiceGroupIdentifier getIdentifier() {
+ return deviceInfo.getServiceIdentifier();
}
- private ListenableFuture<Boolean> chooseStat(final MultipartType multipartType){
- ListenableFuture<Boolean> result = Futures.immediateCheckedFuture(Boolean.TRUE);
-
- switch (multipartType) {
- case OFPMPFLOW:
- result = collectStatistics(multipartType, devState.isFlowStatisticsAvailable(), true);
- break;
- case OFPMPTABLE:
- result = collectStatistics(multipartType, devState.isTableStatisticsAvailable(), false);
- break;
- case OFPMPPORTSTATS:
- result = collectStatistics(multipartType, devState.isPortStatisticsAvailable(), false);
- break;
- case OFPMPQUEUE:
- result = collectStatistics(multipartType, devState.isQueueStatisticsAvailable(), false);
- break;
- case OFPMPGROUPDESC:
- result = collectStatistics(multipartType, devState.isGroupAvailable(), false);
- break;
- case OFPMPGROUP:
- result = collectStatistics(multipartType, devState.isGroupAvailable(), false);
- break;
- case OFPMPMETERCONFIG:
- result = collectStatistics(multipartType, devState.isMetersAvailable(), false);
- break;
- case OFPMPMETER:
- result = collectStatistics(multipartType, devState.isMetersAvailable(), false);
- break;
- default:
- LOG.warn("Unsupported Statistics type {}", multipartType);
- }
-
- return result;
+ @Override
+ public void registerMastershipWatcher(@Nonnull final ContextChainMastershipWatcher newWatcher) {
+ this.contextChainMastershipWatcher = newWatcher;
}
-
@Override
- public <T> RequestContext<T> createRequestContext() {
- final AbstractRequestContext<T> ret = new AbstractRequestContext<T>(deviceInfo.reserveXidForDeviceMessage()) {
+ public <O> RequestContext<O> createRequestContext() {
+ final AbstractRequestContext<O> ret = new AbstractRequestContext<O>(deviceInfo.reserveXidForDeviceMessage()) {
@Override
public void close() {
requestContexts.remove(this);
}
};
+
requestContexts.add(ret);
return ret;
}
@Override
- public void close() {
- if (CONTEXT_STATE.TERMINATION.equals(getState())) {
- if (LOG.isDebugEnabled()) {
- LOG.debug("StatisticsContext for node {} is already in TERMINATION state.", getDeviceInfo().getLOGValue());
- }
- } else {
- this.state = CONTEXT_STATE.TERMINATION;
- stopGatheringData();
-
- for (final Iterator<RequestContext<?>> iterator = Iterators.consumingIterator(requestContexts.iterator());
- iterator.hasNext(); ) {
- RequestContextUtil.closeRequestContextWithRpcError(iterator.next(), CONNECTION_CLOSED);
- }
- }
+ public void enableGathering() {
+ this.schedulingEnabled.set(true);
}
@Override
- public void setSchedulingEnabled(final boolean schedulingEnabled) {
- this.schedulingEnabled = schedulingEnabled;
+ public void disableGathering() {
+ this.schedulingEnabled.set(false);
}
@Override
- public boolean isSchedulingEnabled() {
- return schedulingEnabled;
- }
+ public void continueInitializationAfterReconciliation() {
+ if (deviceContext.initialSubmitTransaction()) {
+ contextChainMastershipWatcher.onMasterRoleAcquired(deviceInfo, ContextChainMastershipState.INITIAL_SUBMIT);
- @Override
- public void setPollTimeout(final Timeout pollTimeout) {
- this.pollTimeout = pollTimeout;
+ startGatheringData();
+ } else {
+ contextChainMastershipWatcher
+ .onNotAbleToStartMastershipMandatory(deviceInfo, "Initial transaction cannot be submitted.");
+ }
}
- private ListenableFuture<Boolean> statChainFuture(final ListenableFuture<Boolean> prevFuture, final MultipartType multipartType) {
- return Futures.transform(deviceConnectionCheck(), (AsyncFunction<Boolean, Boolean>) connectionResult -> Futures
- .transform(prevFuture, (AsyncFunction<Boolean, Boolean>) result -> {
- LOG.debug("Status of previous stat iteration for node {}: {}", deviceInfo.getLOGValue(), result);
- LOG.debug("Stats iterating to next type for node {} of type {}",
- deviceInfo.getLOGValue(),
- multipartType);
-
- return chooseStat(multipartType);
- }));
- }
+ @Override
+ public void instantiateServiceInstance() {
+ final List<MultipartType> statListForCollecting = new ArrayList<>();
- @VisibleForTesting
- ListenableFuture<Boolean> deviceConnectionCheck() {
- if (ConnectionContext.CONNECTION_STATE.RIP.equals(deviceContext.getPrimaryConnectionContext().getConnectionState())) {
- final String errMsg = String
- .format("Device connection for node %s doesn't exist anymore. Primary connection status : %s",
- getDeviceInfo().getNodeId(),
- deviceContext.getPrimaryConnectionContext().getConnectionState());
+ if (devState.isTableStatisticsAvailable()) {
+ statListForCollecting.add(MultipartType.OFPMPTABLE);
+ }
- return Futures.immediateFailedFuture(new ConnectionException(errMsg));
+ if (devState.isFlowStatisticsAvailable()) {
+ statListForCollecting.add(MultipartType.OFPMPFLOW);
}
- return Futures.immediateFuture(Boolean.TRUE);
- }
+ if (devState.isGroupAvailable()) {
+ statListForCollecting.add(MultipartType.OFPMPGROUPDESC);
+ statListForCollecting.add(MultipartType.OFPMPGROUP);
+ }
- private ListenableFuture<Boolean> collectStatistics(final MultipartType multipartType,
- final boolean supported,
- final boolean onTheFly) {
- // TODO: Refactor twice sending deviceContext into gatheringStatistics
- return supported ? StatisticsGatheringUtils.gatherStatistics(
- onTheFly ? statisticsGatheringOnTheFlyService : statisticsGatheringService,
- getDeviceInfo(),
- multipartType,
- deviceContext,
- deviceContext,
- convertorExecutor,
- statisticsWriterProvider) : Futures.immediateFuture(Boolean.FALSE);
- }
+ if (devState.isMetersAvailable()) {
+ statListForCollecting.add(MultipartType.OFPMPMETERCONFIG);
+ statListForCollecting.add(MultipartType.OFPMPMETER);
+ }
- @VisibleForTesting
- void setStatisticsGatheringService(final StatisticsGatheringService<T> statisticsGatheringService) {
- this.statisticsGatheringService = statisticsGatheringService;
- }
+ if (devState.isPortStatisticsAvailable()) {
+ statListForCollecting.add(MultipartType.OFPMPPORTSTATS);
+ }
- @VisibleForTesting
- void setStatisticsGatheringOnTheFlyService(final StatisticsGatheringOnTheFlyService<T> statisticsGatheringOnTheFlyService) {
- this.statisticsGatheringOnTheFlyService = statisticsGatheringOnTheFlyService;
- }
+ if (devState.isQueueStatisticsAvailable()) {
+ statListForCollecting.add(MultipartType.OFPMPQUEUE);
+ }
- @Override
- public ItemLifecycleListener getItemLifeCycleListener () {
- return itemLifeCycleListener;
+ collectingStatType = ImmutableList.copyOf(statListForCollecting);
+ Futures.addCallback(gatherDynamicData(), new InitialSubmitCallback(), MoreExecutors.directExecutor());
}
@Override
- public CONTEXT_STATE getState() {
- return this.state;
+ public ListenableFuture<Void> closeServiceInstance() {
+ return stopGatheringData();
}
@Override
- public ServiceGroupIdentifier getServiceIdentifier() {
- return this.deviceInfo.getServiceIdentifier();
- }
+ public void close() {
+ Futures.addCallback(stopGatheringData(), new FutureCallback<Void>() {
+ @Override
+ public void onSuccess(@Nullable final Void result) {
+ requestContexts.forEach(requestContext -> RequestContextUtil
+ .closeRequestContextWithRpcError(requestContext, CONNECTION_CLOSED));
+ }
- @Override
- public DeviceInfo getDeviceInfo() {
- return this.deviceInfo;
+ @Override
+ public void onFailure(final Throwable throwable) {
+ requestContexts.forEach(requestContext -> RequestContextUtil
+ .closeRequestContextWithRpcError(requestContext, CONNECTION_CLOSED));
+ }
+ }, MoreExecutors.directExecutor());
}
- @Override
- public ListenableFuture<Void> stopClusterServices() {
- if (CONTEXT_STATE.TERMINATION.equals(this.state)) {
- return Futures.immediateCancelledFuture();
+ private ListenableFuture<Boolean> gatherDynamicData() {
+ if (!isStatisticsPollingOn || !schedulingEnabled.get()) {
+ LOG.debug("Statistics for device {} are not enabled.", getDeviceInfo().getNodeId().getValue());
+ return Futures.immediateFuture(Boolean.TRUE);
}
- return Futures.transform(Futures.immediateFuture(null), new Function<Object, Void>() {
- @Nullable
- @Override
- public Void apply(@Nullable Object input) {
- schedulingEnabled = false;
- stopGatheringData();
- return null;
- }
+ return this.lastDataGatheringRef.updateAndGet(future -> {
+ // write start timestamp to state snapshot container
+ StatisticsGatheringUtils.markDeviceStateSnapshotStart(deviceInfo, deviceContext);
+
+ // recreate gathering future if it should be recreated
+ final ListenableFuture<Boolean> lastDataGathering =
+ Objects.isNull(future) || future.isCancelled() || future.isDone() ? Futures
+ .immediateFuture(Boolean.TRUE) : future;
+
+ // build statistics gathering future
+ final ListenableFuture<Boolean> newDataGathering = collectingStatType.stream()
+ .reduce(lastDataGathering, this::statChainFuture,
+ (listenableFuture, asyn) -> Futures.transformAsync(listenableFuture, result -> asyn,
+ MoreExecutors.directExecutor()));
+
+ // write end timestamp to state snapshot container
+ Futures.addCallback(newDataGathering, new FutureCallback<Boolean>() {
+ @Override
+ public void onSuccess(final Boolean result) {
+ StatisticsGatheringUtils.markDeviceStateSnapshotEnd(deviceInfo, deviceContext, result);
+ }
+
+ @Override
+ public void onFailure(final Throwable throwable) {
+ if (!(throwable instanceof TransactionChainClosedException)) {
+ StatisticsGatheringUtils.markDeviceStateSnapshotEnd(deviceInfo, deviceContext, false);
+ }
+ }
+ }, MoreExecutors.directExecutor());
+
+ return newDataGathering;
});
}
- @Override
- public DeviceState gainDeviceState() {
- return gainDeviceContext().getDeviceState();
+ private ListenableFuture<Boolean> statChainFuture(final ListenableFuture<Boolean> prevFuture,
+ final MultipartType multipartType) {
+ if (ConnectionContext.CONNECTION_STATE.RIP
+ .equals(deviceContext.getPrimaryConnectionContext().getConnectionState())) {
+ final String errMsg = String
+ .format("Device connection for node %s doesn't exist anymore. Primary connection status : %s",
+ getDeviceInfo().getNodeId(),
+ deviceContext.getPrimaryConnectionContext().getConnectionState());
+
+ return Futures.immediateFailedFuture(new ConnectionException(errMsg));
+ }
+
+ return Futures.transformAsync(prevFuture, result -> {
+ LOG.debug("Status of previous stat iteration for node {}: {}", deviceInfo, result);
+ LOG.debug("Stats iterating to next type for node {} of type {}", deviceInfo, multipartType);
+ final boolean onTheFly = MultipartType.OFPMPFLOW.equals(multipartType);
+ final boolean supported = collectingStatType.contains(multipartType);
+
+ // TODO: Refactor twice sending deviceContext into gatheringStatistics
+ return supported ? StatisticsGatheringUtils
+ .gatherStatistics(onTheFly ? statisticsGatheringOnTheFlyService : statisticsGatheringService,
+ getDeviceInfo(), multipartType, deviceContext, deviceContext, convertorExecutor,
+ statisticsWriterProvider, executorService) : Futures
+ .immediateFuture(Boolean.FALSE);
+ }, MoreExecutors.directExecutor());
}
- @Override
- public DeviceContext gainDeviceContext() {
- return this.deviceContext;
+ private void startGatheringData() {
+ if (!isStatisticsPollingOn) {
+ return;
+ }
+
+ LOG.info("Starting statistics gathering for node {}", deviceInfo);
+ final StatisticsPollingService statisticsPollingService =
+ new StatisticsPollingService(timeCounter,
+ statisticsPollingInterval,
+ maximumPollingDelay,
+ StatisticsContextImpl.this::gatherDynamicData);
+
+ schedulingEnabled.set(true);
+ statisticsPollingService.startAsync();
+ this.statisticsPollingServiceRef.set(statisticsPollingService);
}
- @Override
- public void stopGatheringData() {
- LOG.info("Stopping running statistics gathering for node {}", deviceInfo.getLOGValue());
+ private ListenableFuture<Void> stopGatheringData() {
+ LOG.info("Stopping running statistics gathering for node {}", deviceInfo);
+ cancelLastDataGathering();
- if (Objects.nonNull(lastDataGathering) && !lastDataGathering.isDone() && !lastDataGathering.isCancelled()) {
- lastDataGathering.cancel(true);
- }
+ return Optional.ofNullable(statisticsPollingServiceRef.getAndSet(null)).map(StatisticsPollingService::stop)
+ .orElseGet(() -> Futures.immediateFuture(null));
+ }
+
+ private void cancelLastDataGathering() {
+ final ListenableFuture<Boolean> future = lastDataGatheringRef.getAndSet(null);
- if (Objects.nonNull(pollTimeout) && !pollTimeout.isExpired()) {
- pollTimeout.cancel();
+ if (Objects.nonNull(future) && !future.isDone() && !future.isCancelled()) {
+ future.cancel(true);
}
}
- @Override
- public void setLifecycleInitializationPhaseHandler(final ClusterInitializationPhaseHandler handler) {
- this.clusterInitializationPhaseHandler = handler;
+ @VisibleForTesting
+ void setStatisticsGatheringService(final StatisticsGatheringService<T> statisticsGatheringService) {
+ this.statisticsGatheringService = statisticsGatheringService;
}
- @Override
- public boolean onContextInstantiateService(final MastershipChangeListener mastershipChangeListener) {
- LOG.info("Starting statistics context cluster services for node {}", deviceInfo.getLOGValue());
- this.statListForCollectingInitialization();
+ @VisibleForTesting
+ void setStatisticsGatheringOnTheFlyService(
+ final StatisticsGatheringOnTheFlyService<T> statisticsGatheringOnTheFlyService) {
+ this.statisticsGatheringOnTheFlyService = statisticsGatheringOnTheFlyService;
+ }
- Futures.addCallback(this.gatherDynamicData(), new FutureCallback<Boolean>() {
- @Override
- public void onSuccess(@Nullable Boolean aBoolean) {
- mastershipChangeListener.onMasterRoleAcquired(
- deviceInfo,
- ContextChainMastershipState.INITIAL_GATHERING
- );
-
- if (initialSubmitHandler.initialSubmitTransaction()) {
- mastershipChangeListener.onMasterRoleAcquired(
- deviceInfo,
- ContextChainMastershipState.INITIAL_SUBMIT
- );
-
- if (isStatisticsPollingOn) {
- myManager.startScheduling(deviceInfo);
- }
- } else {
- mastershipChangeListener.onNotAbleToStartMastershipMandatory(
- deviceInfo,
- "Initial transaction cannot be submitted."
- );
- }
- }
+ private final class InitialSubmitCallback implements FutureCallback<Boolean> {
+ @Override
+ public void onSuccess(@Nullable final Boolean result) {
+ contextChainMastershipWatcher
+ .onMasterRoleAcquired(deviceInfo, ContextChainMastershipState.INITIAL_GATHERING);
- @Override
- public void onFailure(@Nonnull Throwable throwable) {
- mastershipChangeListener.onNotAbleToStartMastershipMandatory(
- deviceInfo,
- "Initial gathering statistics unsuccessful."
- );
+ if (!isUsingReconciliationFramework) {
+ continueInitializationAfterReconciliation();
}
- });
-
- return this.clusterInitializationPhaseHandler.onContextInstantiateService(mastershipChangeListener);
- }
+ }
- @Override
- public void setInitialSubmitHandler(final ClusterInitializationPhaseHandler initialSubmitHandler) {
- this.initialSubmitHandler = initialSubmitHandler;
+ @Override
+ public void onFailure(@Nonnull final Throwable throwable) {
+ contextChainMastershipWatcher.onNotAbleToStartMastershipMandatory(deviceInfo,
+ "Initial gathering statistics "
+ + "unsuccessful: "
+ + throwable.getMessage());
+ }
}
}