2 * Copyright (c) 2016 Pantheon Technologies s.r.o. and others. All rights reserved.
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
8 package org.opendaylight.openflowplugin.impl.lifecycle;
10 import com.google.common.base.Function;
11 import com.google.common.util.concurrent.Futures;
12 import com.google.common.util.concurrent.ListenableFuture;
13 import io.netty.util.internal.ConcurrentSet;
14 import java.util.ArrayList;
15 import java.util.List;
17 import java.util.concurrent.atomic.AtomicBoolean;
18 import javax.annotation.Nonnull;
19 import javax.annotation.Nullable;
20 import org.opendaylight.mdsal.singleton.common.api.ClusterSingletonServiceProvider;
21 import org.opendaylight.openflowplugin.api.openflow.OFPContext;
22 import org.opendaylight.openflowplugin.api.openflow.connection.ConnectionContext;
23 import org.opendaylight.openflowplugin.api.openflow.device.DeviceContext;
24 import org.opendaylight.openflowplugin.api.openflow.device.DeviceInfo;
25 import org.opendaylight.openflowplugin.api.openflow.lifecycle.ContextChain;
26 import org.opendaylight.openflowplugin.api.openflow.lifecycle.ContextChainMastershipState;
27 import org.opendaylight.openflowplugin.api.openflow.lifecycle.ContextChainState;
28 import org.opendaylight.openflowplugin.api.openflow.lifecycle.ContextChainStateListener;
29 import org.opendaylight.openflowplugin.api.openflow.lifecycle.LifecycleService;
30 import org.opendaylight.openflowplugin.api.openflow.rpc.RpcContext;
31 import org.opendaylight.openflowplugin.api.openflow.statistics.StatisticsContext;
32 import org.slf4j.Logger;
33 import org.slf4j.LoggerFactory;
35 public class ContextChainImpl implements ContextChain {
37 private static final Logger LOG = LoggerFactory.getLogger(ContextChainImpl.class);
39 private Set<OFPContext> contexts = new ConcurrentSet<>();
40 private StatisticsContext statisticsContext;
41 private DeviceContext deviceContext;
42 private RpcContext rpcContext;
43 private LifecycleService lifecycleService;
44 private DeviceInfo deviceInfo;
45 private ConnectionContext primaryConnection;
46 private Set<ConnectionContext> auxiliaryConnections = new ConcurrentSet<>();
48 private volatile ContextChainState contextChainState = ContextChainState.UNDEFINED;
50 private final AtomicBoolean masterStateOnDevice = new AtomicBoolean(false);
51 private final AtomicBoolean initialGathering = new AtomicBoolean(false);
52 private final AtomicBoolean initialSubmitting = new AtomicBoolean(false);
53 private final AtomicBoolean registryFilling = new AtomicBoolean(false);
54 private final AtomicBoolean rpcRegistration = new AtomicBoolean(false);
56 ContextChainImpl(final ConnectionContext connectionContext) {
57 this.primaryConnection = connectionContext;
58 this.deviceInfo = connectionContext.getDeviceInfo();
62 public <T extends OFPContext> void addContext(final T context) {
63 if (context instanceof StatisticsContext) {
64 this.statisticsContext = (StatisticsContext) context;
66 if (context instanceof DeviceContext) {
67 this.deviceContext = (DeviceContext) context;
69 if (context instanceof RpcContext) {
70 this.rpcContext = (RpcContext) context;
75 contexts.add(context);
79 public void addLifecycleService(final LifecycleService lifecycleService) {
80 this.lifecycleService = lifecycleService;
84 public ListenableFuture<Void> stopChain() {
85 final List<ListenableFuture<Void>> futureList = new ArrayList<>();
86 futureList.add(statisticsContext.stopClusterServices());
87 futureList.add(rpcContext.stopClusterServices());
88 futureList.add(deviceContext.stopClusterServices());
90 return Futures.transform(Futures.successfulAsList(futureList), new Function<List<Void>, Void>() {
93 public Void apply(@Nullable List<Void> input) {
94 LOG.info("Closed clustering MASTER services for node {}", deviceContext.getDeviceInfo().getLOGValue());
100 private void unMasterMe() {
101 this.registryFilling.set(false);
102 this.initialSubmitting.set(false);
103 this.initialGathering.set(false);
104 this.masterStateOnDevice.set(false);
105 this.rpcRegistration.set(false);
109 public void close() {
110 this.auxiliaryConnections.forEach(connectionContext -> connectionContext.closeConnection(false));
111 this.primaryConnection.closeConnection(true);
112 lifecycleService.close();
113 deviceContext.close();
115 statisticsContext.close();
120 public void makeContextChainStateSlave() {
122 changeState(ContextChainState.WORKING_SLAVE);
126 public ListenableFuture<Void> connectionDropped() {
127 if (this.contextChainState == ContextChainState.WORKING_MASTER) {
128 return this.stopChain();
131 return Futures.immediateFuture(null);
135 public void registerServices(final ClusterSingletonServiceProvider clusterSingletonServiceProvider) {
136 this.lifecycleService.registerService(
137 clusterSingletonServiceProvider,
142 public void makeDeviceSlave() {
144 this.lifecycleService.makeDeviceSlave(this.deviceContext);
148 public boolean isMastered(@Nonnull ContextChainMastershipState mastershipState) {
149 switch (mastershipState) {
151 LOG.debug("Device {}, initial submit OK.", deviceInfo.getLOGValue());
152 this.initialSubmitting.set(true);
154 case MASTER_ON_DEVICE:
155 LOG.debug("Device {}, master state OK.", deviceInfo.getLOGValue());
156 this.masterStateOnDevice.set(true);
158 case INITIAL_GATHERING:
159 LOG.debug("Device {}, initial gathering OK.", deviceInfo.getLOGValue());
160 this.initialGathering.set(true);
162 case RPC_REGISTRATION:
163 LOG.debug("Device {}, RPC registration OK.", deviceInfo.getLOGValue());
164 this.rpcRegistration.set(true);
165 //Flow registry fill is not mandatory to work as a master
166 case INITIAL_FLOW_REGISTRY_FILL:
167 LOG.debug("Device {}, initial registry filling OK.", deviceInfo.getLOGValue());
168 this.registryFilling.set(true);
173 final boolean result =
174 this.initialGathering.get() &&
175 this.masterStateOnDevice.get() &&
176 this.initialSubmitting.get() &&
177 this.rpcRegistration.get();
179 if (result && mastershipState != ContextChainMastershipState.CHECK) {
180 LOG.info("Device {} is able to work as master{}",
181 deviceInfo.getLOGValue(),
182 this.registryFilling.get() ? "." : " WITHOUT flow registry !!!");
183 changeState(ContextChainState.WORKING_MASTER);
190 public boolean addAuxiliaryConnection(@Nonnull ConnectionContext connectionContext) {
191 if ((connectionContext.getFeatures().getAuxiliaryId() != 0) &&
192 (this.primaryConnection.getConnectionState() != ConnectionContext.CONNECTION_STATE.RIP)) {
193 this.auxiliaryConnections.add(connectionContext);
201 public boolean auxiliaryConnectionDropped(@Nonnull ConnectionContext connectionContext) {
202 if (this.auxiliaryConnections.isEmpty()) {
205 if (!this.auxiliaryConnections.contains(connectionContext)) {
208 this.auxiliaryConnections.remove(connectionContext);
212 private void changeState(final ContextChainState contextChainState) {
213 boolean propagate = this.contextChainState == ContextChainState.UNDEFINED;
214 this.contextChainState = contextChainState;
218 .filter(ContextChainStateListener.class::isInstance)
219 .map(ContextChainStateListener.class::cast)
220 .forEach(listener -> listener.onStateAcquired(contextChainState));