2 * Copyright (c) 2013 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
10 package org.opendaylight.openflowjava.protocol.impl.core.connection;
12 import com.google.common.base.Preconditions;
13 import com.google.common.cache.Cache;
14 import com.google.common.cache.CacheBuilder;
15 import com.google.common.cache.RemovalCause;
16 import com.google.common.cache.RemovalListener;
17 import com.google.common.cache.RemovalNotification;
18 import com.google.common.util.concurrent.ListenableFuture;
19 import com.google.common.util.concurrent.SettableFuture;
20 import io.netty.channel.Channel;
21 import io.netty.channel.ChannelFuture;
22 import io.netty.util.concurrent.GenericFutureListener;
23 import java.net.InetSocketAddress;
24 import java.util.concurrent.Future;
25 import java.util.concurrent.RejectedExecutionException;
26 import java.util.concurrent.TimeUnit;
27 import org.opendaylight.openflowjava.protocol.api.connection.ConnectionReadyListener;
28 import org.opendaylight.openflowjava.protocol.api.connection.OutboundQueueHandler;
29 import org.opendaylight.openflowjava.protocol.api.connection.OutboundQueueHandlerRegistration;
30 import org.opendaylight.openflowjava.protocol.impl.core.OFVersionDetector;
31 import org.opendaylight.openflowjava.protocol.impl.core.PipelineHandlers;
32 import org.opendaylight.openflowjava.statistics.CounterEventTypes;
33 import org.opendaylight.openflowjava.statistics.StatisticsCounters;
34 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.BarrierInput;
35 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.BarrierOutput;
36 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.EchoInput;
37 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.EchoOutput;
38 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.EchoReplyInput;
39 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.EchoRequestMessage;
40 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.ErrorMessage;
41 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.ExperimenterInput;
42 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.ExperimenterMessage;
43 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.FlowModInput;
44 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.FlowRemovedMessage;
45 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.GetAsyncInput;
46 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.GetAsyncOutput;
47 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.GetConfigInput;
48 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.GetConfigOutput;
49 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.GetFeaturesInput;
50 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.GetFeaturesOutput;
51 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.GetQueueConfigInput;
52 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.GetQueueConfigOutput;
53 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.GroupModInput;
54 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.HelloInput;
55 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.HelloMessage;
56 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.MeterModInput;
57 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.MultipartReplyMessage;
58 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.MultipartRequestInput;
59 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.OfHeader;
60 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.OpenflowProtocolListener;
61 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.PacketInMessage;
62 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.PacketOutInput;
63 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.PortModInput;
64 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.PortStatusMessage;
65 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.RoleRequestInput;
66 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.RoleRequestOutput;
67 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.SetAsyncInput;
68 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.SetConfigInput;
69 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.TableModInput;
70 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.system.rev130927.DisconnectEvent;
71 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.system.rev130927.SwitchIdleEvent;
72 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.system.rev130927.SystemNotificationsListener;
73 import org.opendaylight.yangtools.yang.binding.DataObject;
74 import org.opendaylight.yangtools.yang.binding.Notification;
75 import org.opendaylight.yangtools.yang.common.RpcResult;
76 import org.slf4j.Logger;
77 import org.slf4j.LoggerFactory;
80 * Handles messages (notifications + rpcs) and connections
82 * @author michal.polkorab
84 public class ConnectionAdapterImpl implements ConnectionFacade {
85 /** after this time, RPC future response objects will be thrown away (in minutes) */
86 public static final int RPC_RESPONSE_EXPIRATION = 1;
89 * Default depth of write queue, e.g. we allow these many messages
90 * to be queued up before blocking producers.
92 public static final int DEFAULT_QUEUE_DEPTH = 1024;
94 private static final Logger LOG = LoggerFactory
95 .getLogger(ConnectionAdapterImpl.class);
96 private static final Exception QUEUE_FULL_EXCEPTION =
97 new RejectedExecutionException("Output queue is full");
99 private static final RemovalListener<RpcResponseKey, ResponseExpectedRpcListener<?>> REMOVAL_LISTENER =
100 new RemovalListener<RpcResponseKey, ResponseExpectedRpcListener<?>>() {
102 public void onRemoval(
103 final RemovalNotification<RpcResponseKey, ResponseExpectedRpcListener<?>> notification) {
104 if (! notification.getCause().equals(RemovalCause.EXPLICIT)) {
105 notification.getValue().discard();
110 /** expiring cache for future rpcResponses */
111 private Cache<RpcResponseKey, ResponseExpectedRpcListener<?>> responseCache;
113 private final ChannelOutboundQueue output;
114 private final Channel channel;
116 private ConnectionReadyListener connectionReadyListener;
117 private OpenflowProtocolListener messageListener;
118 private SystemNotificationsListener systemListener;
119 private OutboundQueueManager<?> outputManager;
120 private boolean disconnectOccured = false;
121 private final StatisticsCounters statisticsCounters;
122 private OFVersionDetector versionDetector;
123 private final InetSocketAddress address;
127 * @param channel the channel to be set - used for communication
128 * @param address client address (used only in case of UDP communication,
129 * as there is no need to store address over tcp (stable channel))
131 public ConnectionAdapterImpl(final Channel channel, final InetSocketAddress address) {
132 responseCache = CacheBuilder.newBuilder()
134 .expireAfterWrite(RPC_RESPONSE_EXPIRATION, TimeUnit.MINUTES)
135 .removalListener(REMOVAL_LISTENER).build();
136 this.channel = Preconditions.checkNotNull(channel);
137 this.output = new ChannelOutboundQueue(channel, DEFAULT_QUEUE_DEPTH, address);
138 this.address = address;
139 channel.pipeline().addLast(output);
140 statisticsCounters = StatisticsCounters.getInstance();
142 LOG.debug("ConnectionAdapter created");
146 public Future<RpcResult<BarrierOutput>> barrier(final BarrierInput input) {
147 return sendToSwitchExpectRpcResultFuture(
148 input, BarrierOutput.class, "barrier-input sending failed");
152 public Future<RpcResult<EchoOutput>> echo(final EchoInput input) {
153 return sendToSwitchExpectRpcResultFuture(
154 input, EchoOutput.class, "echo-input sending failed");
158 public Future<RpcResult<Void>> echoReply(final EchoReplyInput input) {
159 return sendToSwitchFuture(input, "echo-reply sending failed");
163 public Future<RpcResult<Void>> experimenter(final ExperimenterInput input) {
164 return sendToSwitchFuture(input, "experimenter sending failed");
168 public Future<RpcResult<Void>> flowMod(final FlowModInput input) {
169 statisticsCounters.incrementCounter(CounterEventTypes.DS_FLOW_MODS_ENTERED);
170 return sendToSwitchFuture(input, "flow-mod sending failed");
174 public Future<RpcResult<GetConfigOutput>> getConfig(final GetConfigInput input) {
175 return sendToSwitchExpectRpcResultFuture(
176 input, GetConfigOutput.class, "get-config-input sending failed");
180 public Future<RpcResult<GetFeaturesOutput>> getFeatures(
181 final GetFeaturesInput input) {
182 return sendToSwitchExpectRpcResultFuture(
183 input, GetFeaturesOutput.class, "get-features-input sending failed");
187 public Future<RpcResult<GetQueueConfigOutput>> getQueueConfig(
188 final GetQueueConfigInput input) {
189 return sendToSwitchExpectRpcResultFuture(
190 input, GetQueueConfigOutput.class, "get-queue-config-input sending failed");
194 public Future<RpcResult<Void>> groupMod(final GroupModInput input) {
195 return sendToSwitchFuture(input, "group-mod-input sending failed");
199 public Future<RpcResult<Void>> hello(final HelloInput input) {
200 return sendToSwitchFuture(input, "hello-input sending failed");
204 public Future<RpcResult<Void>> meterMod(final MeterModInput input) {
205 return sendToSwitchFuture(input, "meter-mod-input sending failed");
209 public Future<RpcResult<Void>> packetOut(final PacketOutInput input) {
210 return sendToSwitchFuture(input, "packet-out-input sending failed");
214 public Future<RpcResult<Void>> multipartRequest(final MultipartRequestInput input) {
215 return sendToSwitchFuture(input, "multi-part-request sending failed");
219 public Future<RpcResult<Void>> portMod(final PortModInput input) {
220 return sendToSwitchFuture(input, "port-mod-input sending failed");
224 public Future<RpcResult<RoleRequestOutput>> roleRequest(
225 final RoleRequestInput input) {
226 return sendToSwitchExpectRpcResultFuture(
227 input, RoleRequestOutput.class, "role-request-config-input sending failed");
231 public Future<RpcResult<Void>> setConfig(final SetConfigInput input) {
232 return sendToSwitchFuture(input, "set-config-input sending failed");
236 public Future<RpcResult<Void>> tableMod(final TableModInput input) {
237 return sendToSwitchFuture(input, "table-mod-input sending failed");
241 public Future<RpcResult<GetAsyncOutput>> getAsync(final GetAsyncInput input) {
242 return sendToSwitchExpectRpcResultFuture(
243 input, GetAsyncOutput.class, "get-async-input sending failed");
247 public Future<RpcResult<Void>> setAsync(final SetAsyncInput input) {
248 return sendToSwitchFuture(input, "set-async-input sending failed");
252 public Future<Boolean> disconnect() {
253 ChannelFuture disconnectResult = channel.disconnect();
254 responseCache.invalidateAll();
255 disconnectOccured = true;
257 return handleTransportChannelFuture(disconnectResult);
261 public boolean isAlive() {
262 return channel.isOpen();
266 public void setMessageListener(final OpenflowProtocolListener messageListener) {
267 this.messageListener = messageListener;
271 public void consume(final DataObject message) {
272 LOG.debug("ConsumeIntern msg on {}", channel);
273 if (disconnectOccured ) {
276 if (message instanceof Notification) {
279 if (message instanceof DisconnectEvent) {
280 systemListener.onDisconnectEvent((DisconnectEvent) message);
281 responseCache.invalidateAll();
282 disconnectOccured = true;
283 } else if (message instanceof SwitchIdleEvent) {
284 systemListener.onSwitchIdleEvent((SwitchIdleEvent) message);
286 } else if (message instanceof EchoRequestMessage) {
287 if (outputManager != null) {
288 outputManager.onEchoRequest((EchoRequestMessage) message);
290 messageListener.onEchoRequestMessage((EchoRequestMessage) message);
292 statisticsCounters.incrementCounter(CounterEventTypes.US_MESSAGE_PASS);
293 } else if (message instanceof ErrorMessage) {
294 // Send only unmatched errors
295 if (outputManager == null || !outputManager.onMessage((OfHeader) message)) {
296 messageListener.onErrorMessage((ErrorMessage) message);
298 statisticsCounters.incrementCounter(CounterEventTypes.US_MESSAGE_PASS);
299 } else if (message instanceof ExperimenterMessage) {
300 if (outputManager != null) {
301 outputManager.onMessage((OfHeader) message);
303 messageListener.onExperimenterMessage((ExperimenterMessage) message);
304 statisticsCounters.incrementCounter(CounterEventTypes.US_MESSAGE_PASS);
305 } else if (message instanceof FlowRemovedMessage) {
306 messageListener.onFlowRemovedMessage((FlowRemovedMessage) message);
307 statisticsCounters.incrementCounter(CounterEventTypes.US_MESSAGE_PASS);
308 } else if (message instanceof HelloMessage) {
309 LOG.info("Hello received / branch");
310 messageListener.onHelloMessage((HelloMessage) message);
311 statisticsCounters.incrementCounter(CounterEventTypes.US_MESSAGE_PASS);
312 } else if (message instanceof MultipartReplyMessage) {
313 if (outputManager != null) {
314 outputManager.onMessage((OfHeader) message);
316 messageListener.onMultipartReplyMessage((MultipartReplyMessage) message);
317 statisticsCounters.incrementCounter(CounterEventTypes.US_MESSAGE_PASS);
318 } else if (message instanceof PacketInMessage) {
319 messageListener.onPacketInMessage((PacketInMessage) message);
320 statisticsCounters.incrementCounter(CounterEventTypes.US_MESSAGE_PASS);
321 } else if (message instanceof PortStatusMessage) {
322 messageListener.onPortStatusMessage((PortStatusMessage) message);
323 statisticsCounters.incrementCounter(CounterEventTypes.US_MESSAGE_PASS);
325 LOG.warn("message listening not supported for type: {}", message.getClass());
327 } else if (message instanceof OfHeader) {
328 LOG.debug("OFheader msg received");
330 if (outputManager == null || !outputManager.onMessage((OfHeader) message)) {
331 RpcResponseKey key = createRpcResponseKey((OfHeader) message);
332 final ResponseExpectedRpcListener<?> listener = findRpcResponse(key);
333 if (listener != null) {
334 LOG.debug("corresponding rpcFuture found");
335 listener.completed((OfHeader)message);
336 statisticsCounters.incrementCounter(CounterEventTypes.US_MESSAGE_PASS);
337 LOG.debug("after setting rpcFuture");
338 responseCache.invalidate(key);
340 LOG.warn("received unexpected rpc response: {}", key);
344 LOG.warn("message listening not supported for type: {}", message.getClass());
348 private <T> ListenableFuture<RpcResult<T>> enqueueMessage(final AbstractRpcListener<T> promise) {
349 LOG.debug("Submitting promise {}", promise);
351 if (!output.enqueue(promise)) {
352 LOG.debug("Message queue is full, rejecting execution");
353 promise.failedRpc(QUEUE_FULL_EXCEPTION);
355 LOG.debug("Promise enqueued successfully");
358 return promise.getResult();
362 * sends given message to switch, sending result will be reported via return value
363 * @param input message to send
364 * @param failureInfo describes, what type of message caused failure by sending
365 * @return future object, <ul>
366 * <li>if send successful, {@link RpcResult} without errors and successful
367 * status will be returned, </li>
368 * <li>else {@link RpcResult} will contain errors and failed status</li>
371 private ListenableFuture<RpcResult<Void>> sendToSwitchFuture(
372 final DataObject input, final String failureInfo) {
373 statisticsCounters.incrementCounter(CounterEventTypes.DS_ENTERED_OFJAVA);
374 return enqueueMessage(new SimpleRpcListener(input, failureInfo));
378 * sends given message to switch, sending result or switch response will be reported via return value
379 * @param input message to send
380 * @param responseClazz type of response
381 * @param failureInfo describes, what type of message caused failure by sending
382 * @return future object, <ul>
383 * <li>if send fails, {@link RpcResult} will contain errors and failed status </li>
384 * <li>else {@link RpcResult} will be stored in responseCache and wait for particular timeout
385 * ({@link ConnectionAdapterImpl#RPC_RESPONSE_EXPIRATION}),
386 * <ul><li>either switch will manage to answer
387 * and then corresponding response message will be set into returned future</li>
388 * <li>or response in cache will expire and returned future will be cancelled</li></ul>
392 private <IN extends OfHeader, OUT extends OfHeader> ListenableFuture<RpcResult<OUT>> sendToSwitchExpectRpcResultFuture(
393 final IN input, final Class<OUT> responseClazz, final String failureInfo) {
394 final RpcResponseKey key = new RpcResponseKey(input.getXid(), responseClazz.getName());
395 final ResponseExpectedRpcListener<OUT> listener =
396 new ResponseExpectedRpcListener<>(input, failureInfo, responseCache, key);
397 statisticsCounters.incrementCounter(CounterEventTypes.DS_ENTERED_OFJAVA);
398 return enqueueMessage(listener);
402 * @param resultFuture
404 * @param errorSeverity
408 private static SettableFuture<Boolean> handleTransportChannelFuture(
409 final ChannelFuture resultFuture) {
411 final SettableFuture<Boolean> transportResult = SettableFuture.create();
413 resultFuture.addListener(new GenericFutureListener<io.netty.util.concurrent.Future<? super Void>>() {
416 public void operationComplete(
417 final io.netty.util.concurrent.Future<? super Void> future)
419 transportResult.set(future.isSuccess());
420 if (!future.isSuccess()) {
421 transportResult.setException(future.cause());
425 return transportResult;
432 private static RpcResponseKey createRpcResponseKey(final OfHeader message) {
433 return new RpcResponseKey(message.getXid(), message.getImplementedInterface().getName());
439 private ResponseExpectedRpcListener<?> findRpcResponse(final RpcResponseKey key) {
440 return responseCache.getIfPresent(key);
444 public void setSystemListener(final SystemNotificationsListener systemListener) {
445 this.systemListener = systemListener;
449 public void checkListeners() {
450 final StringBuilder buffer = new StringBuilder();
451 if (systemListener == null) {
452 buffer.append("SystemListener ");
454 if (messageListener == null) {
455 buffer.append("MessageListener ");
457 if (connectionReadyListener == null) {
458 buffer.append("ConnectionReadyListener ");
461 Preconditions.checkState(buffer.length() == 0, "Missing listeners: %s", buffer.toString());
465 public void fireConnectionReadyNotification() {
466 versionDetector = (OFVersionDetector) channel.pipeline().get(PipelineHandlers.OF_VERSION_DETECTOR.name());
467 Preconditions.checkState(versionDetector != null);
469 new Thread(new Runnable() {
472 connectionReadyListener.onConnectionReady();
478 public void setConnectionReadyListener(
479 final ConnectionReadyListener connectionReadyListener) {
480 this.connectionReadyListener = connectionReadyListener;
484 public InetSocketAddress getRemoteAddress() {
485 return (InetSocketAddress) channel.remoteAddress();
489 * Used only for testing purposes
492 public void setResponseCache(final Cache<RpcResponseKey, ResponseExpectedRpcListener<?>> cache) {
493 this.responseCache = cache;
497 public boolean isAutoRead() {
498 return channel.config().isAutoRead();
502 public void setAutoRead(final boolean autoRead) {
503 channel.config().setAutoRead(autoRead);
507 public <T extends OutboundQueueHandler> OutboundQueueHandlerRegistration<T> registerOutboundQueueHandler(
508 final T handler, final int maxQueueDepth, final long maxBarrierNanos) {
509 Preconditions.checkState(outputManager == null, "Manager %s already registered", outputManager);
511 final OutboundQueueManager<T> ret = new OutboundQueueManager<>(this, address, handler, OutboundQueueCache.getInstance().getSlice(maxQueueDepth), maxBarrierNanos);
513 channel.pipeline().addLast(outputManager);
515 return new OutboundQueueHandlerRegistrationImpl<T>(handler) {
517 protected void removeRegistration() {
518 outputManager.close();
519 channel.pipeline().remove(outputManager);
520 outputManager = null;
525 Channel getChannel() {
530 public void setPacketInFiltering(final boolean enabled) {
531 versionDetector.setFilterPacketIns(enabled);
532 LOG.debug("PacketIn filtering {}abled", enabled ? "en" : "dis");