2 * Copyright (c) 2015 Cisco Systems, Inc. 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.connection;
10 import static java.util.Objects.requireNonNull;
12 import com.google.common.annotations.VisibleForTesting;
13 import com.google.common.util.concurrent.ThreadFactoryBuilder;
14 import java.math.BigInteger;
15 import java.net.InetAddress;
16 import java.time.LocalDateTime;
17 import java.util.Collection;
19 import java.util.concurrent.ConcurrentHashMap;
20 import java.util.concurrent.ExecutorService;
21 import java.util.concurrent.Executors;
22 import java.util.concurrent.ThreadFactory;
23 import org.eclipse.jdt.annotation.NonNull;
24 import org.opendaylight.mdsal.binding.api.ClusteredDataTreeChangeListener;
25 import org.opendaylight.mdsal.binding.api.DataBroker;
26 import org.opendaylight.mdsal.binding.api.DataObjectModification;
27 import org.opendaylight.mdsal.binding.api.DataTreeIdentifier;
28 import org.opendaylight.mdsal.binding.api.DataTreeModification;
29 import org.opendaylight.mdsal.binding.api.NotificationPublishService;
30 import org.opendaylight.mdsal.common.api.LogicalDatastoreType;
31 import org.opendaylight.openflowjava.protocol.api.connection.ConnectionAdapter;
32 import org.opendaylight.openflowjava.protocol.api.connection.ConnectionReadyListener;
33 import org.opendaylight.openflowplugin.api.OFConstants;
34 import org.opendaylight.openflowplugin.api.openflow.connection.ConnectionContext;
35 import org.opendaylight.openflowplugin.api.openflow.connection.ConnectionManager;
36 import org.opendaylight.openflowplugin.api.openflow.connection.DeviceConnectionStatusProvider;
37 import org.opendaylight.openflowplugin.api.openflow.connection.HandshakeContext;
38 import org.opendaylight.openflowplugin.api.openflow.device.handlers.DeviceConnectedHandler;
39 import org.opendaylight.openflowplugin.api.openflow.device.handlers.DeviceDisconnectedHandler;
40 import org.opendaylight.openflowplugin.api.openflow.md.core.HandshakeListener;
41 import org.opendaylight.openflowplugin.api.openflow.md.core.HandshakeManager;
42 import org.opendaylight.openflowplugin.impl.common.DeviceConnectionRateLimiter;
43 import org.opendaylight.openflowplugin.impl.connection.listener.ConnectionReadyListenerImpl;
44 import org.opendaylight.openflowplugin.impl.connection.listener.HandshakeListenerImpl;
45 import org.opendaylight.openflowplugin.impl.connection.listener.OpenflowProtocolListenerInitialImpl;
46 import org.opendaylight.openflowplugin.impl.connection.listener.SystemNotificationsListenerImpl;
47 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.Nodes;
48 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node;
49 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.openflow.provider.config.rev160510.OpenflowProviderConfig;
50 import org.opendaylight.yangtools.concepts.ListenerRegistration;
51 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
52 import org.slf4j.Logger;
53 import org.slf4j.LoggerFactory;
55 public class ConnectionManagerImpl implements ConnectionManager {
57 private static final Logger LOG = LoggerFactory.getLogger(ConnectionManagerImpl.class);
58 private static final Logger OF_EVENT_LOG = LoggerFactory.getLogger("OfEventLog");
59 private static final boolean BITMAP_NEGOTIATION_ENABLED = true;
60 private final ThreadFactory threadFactory = new ThreadFactoryBuilder()
61 .setNameFormat("ConnectionHandler-%d")
63 .setUncaughtExceptionHandler((thread, ex) -> LOG.error("Uncaught exception {}", thread, ex))
65 private final ExecutorService executorsService = Executors.newCachedThreadPool(threadFactory);
67 private DeviceConnectedHandler deviceConnectedHandler;
68 private final OpenflowProviderConfig config;
69 private final ExecutorService executorService;
70 private final DeviceConnectionRateLimiter deviceConnectionRateLimiter;
71 private final DataBroker dataBroker;
72 private final int deviceConnectionHoldTime;
73 private DeviceDisconnectedHandler deviceDisconnectedHandler;
74 private DeviceConnectionStatusProvider deviceConnectionStatusProvider;
75 private final NotificationPublishService notificationPublishService;
77 public ConnectionManagerImpl(final OpenflowProviderConfig config, final ExecutorService executorService,
78 final DataBroker dataBroker,
79 @NonNull final NotificationPublishService notificationPublishService) {
81 this.executorService = executorService;
82 deviceConnectionRateLimiter = new DeviceConnectionRateLimiter(config);
83 this.dataBroker = dataBroker;
84 deviceConnectionHoldTime = config.getDeviceConnectionHoldTimeInSeconds().toJava();
85 deviceConnectionStatusProvider = new DeviceConnectionStatusProviderImpl();
86 deviceConnectionStatusProvider.init();
87 this.notificationPublishService = notificationPublishService;
91 public void onSwitchConnected(final ConnectionAdapter connectionAdapter) {
92 connectionAdapter.setExecutorService(executorsService);
93 OF_EVENT_LOG.debug("OnSwitchConnected event received for device {}", connectionAdapter.getRemoteAddress());
94 LOG.trace("prepare connection context");
95 final ConnectionContext connectionContext = new ConnectionContextImpl(connectionAdapter,
96 deviceConnectionStatusProvider);
97 connectionContext.setDeviceDisconnectedHandler(deviceDisconnectedHandler);
99 HandshakeListener handshakeListener = new HandshakeListenerImpl(connectionContext, deviceConnectedHandler);
100 final HandshakeManager handshakeManager = createHandshakeManager(connectionAdapter, handshakeListener);
102 LOG.trace("prepare handshake context");
103 HandshakeContext handshakeContext = new HandshakeContextImpl(executorService, handshakeManager);
104 handshakeListener.setHandshakeContext(handshakeContext);
105 connectionContext.setHandshakeContext(handshakeContext);
107 LOG.trace("prepare connection listeners");
108 final ConnectionReadyListener connectionReadyListener = new ConnectionReadyListenerImpl(
109 connectionContext, handshakeContext);
110 connectionAdapter.setConnectionReadyListener(connectionReadyListener);
112 connectionAdapter.setMessageListener(
113 new OpenflowProtocolListenerInitialImpl(connectionContext, handshakeContext));
115 connectionAdapter.setSystemListener(new SystemNotificationsListenerImpl(connectionContext,
116 config.getEchoReplyTimeout().getValue().toJava(), executorService, notificationPublishService));
118 LOG.trace("connection ballet finished");
121 private HandshakeManager createHandshakeManager(final ConnectionAdapter connectionAdapter,
122 final HandshakeListener handshakeListener) {
123 HandshakeManagerImpl handshakeManager = new HandshakeManagerImpl(connectionAdapter,
124 OFConstants.VERSION_ORDER.get(0),
125 OFConstants.VERSION_ORDER, new ErrorHandlerSimpleImpl(), handshakeListener, BITMAP_NEGOTIATION_ENABLED,
126 deviceConnectionRateLimiter, deviceConnectionHoldTime, deviceConnectionStatusProvider);
128 return handshakeManager;
132 public boolean accept(final InetAddress switchAddress) {
133 // TODO add connection accept logic based on address
138 public void setDeviceConnectedHandler(final DeviceConnectedHandler deviceConnectedHandler) {
139 this.deviceConnectedHandler = deviceConnectedHandler;
143 public void setDeviceDisconnectedHandler(final DeviceDisconnectedHandler deviceDisconnectedHandler) {
144 this.deviceDisconnectedHandler = deviceDisconnectedHandler;
148 DeviceConnectionStatusProvider getDeviceConnectionStatusProvider() {
149 return deviceConnectionStatusProvider;
153 public void close() throws Exception {
154 if (deviceConnectionStatusProvider != null) {
155 deviceConnectionStatusProvider.close();
156 deviceConnectionStatusProvider = null;
158 if (executorsService != null) {
159 executorsService.shutdownNow();
163 class DeviceConnectionStatusProviderImpl implements DeviceConnectionStatusProvider,
164 ClusteredDataTreeChangeListener<Node> {
165 private final Map<BigInteger, LocalDateTime> deviceConnectionMap = new ConcurrentHashMap<>();
167 private ListenerRegistration<DeviceConnectionStatusProviderImpl> listenerRegistration;
170 @SuppressWarnings({"checkstyle:IllegalCatch"})
172 DataTreeIdentifier<Node> treeId = DataTreeIdentifier.create(LogicalDatastoreType.OPERATIONAL,
175 listenerRegistration = dataBroker.registerDataTreeChangeListener(treeId, this);
176 } catch (Exception e) {
177 LOG.error("DeviceConnectionStatusProvider listener registration failed", e);
182 public LocalDateTime getDeviceLastConnectionTime(final BigInteger nodeId) {
183 return deviceConnectionMap.get(nodeId);
187 public void addDeviceLastConnectionTime(final BigInteger nodeId, final LocalDateTime time) {
188 deviceConnectionMap.put(nodeId, time);
192 public void removeDeviceLastConnectionTime(final BigInteger nodeId) {
193 deviceConnectionMap.remove(nodeId);
197 public void onDataTreeChanged(@NonNull final Collection<DataTreeModification<Node>> changes) {
198 requireNonNull(changes, "Changes must not be null!");
199 for (DataTreeModification<Node> change : changes) {
200 final DataObjectModification<Node> mod = change.getRootNode();
201 switch (mod.getModificationType()) {
204 case SUBTREE_MODIFIED:
207 processNodeModification(change);
210 throw new IllegalArgumentException("Unhandled modification type " + mod.getModificationType());
215 private InstanceIdentifier<Node> getWildCardPath() {
216 return InstanceIdentifier.create(Nodes.class).child(Node.class);
219 private void processNodeModification(final DataTreeModification<Node> change) {
220 final InstanceIdentifier<Node> key = change.getRootPath().getRootIdentifier();
221 final InstanceIdentifier<Node> nodeIdent = key.firstIdentifierOf(Node.class);
222 String[] nodeIdentity = nodeIdent.firstKeyOf(Node.class).getId().getValue().split(":");
223 String nodeId = nodeIdentity[1];
224 LOG.info("Clearing the device connection timer for the device {}", nodeId);
225 removeDeviceLastConnectionTime(new BigInteger(nodeId));
229 public void close() {
230 if (listenerRegistration != null) {
231 listenerRegistration.close();
232 listenerRegistration = null;