Merge "BUG-1511 - Datastore: cleanup ListenerTreeAPI"
[controller.git] / opendaylight / md-sal / sal-netconf-connector / src / main / java / org / opendaylight / controller / sal / connect / netconf / listener / NetconfDeviceCommunicator.java
1 /*
2  * Copyright (c) 2014 Cisco Systems, Inc. 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.sal.connect.netconf.listener;
9
10 import com.google.common.base.Optional;
11 import com.google.common.base.Strings;
12 import com.google.common.collect.Lists;
13 import com.google.common.util.concurrent.Futures;
14 import com.google.common.util.concurrent.ListenableFuture;
15 import io.netty.util.concurrent.Future;
16 import io.netty.util.concurrent.FutureListener;
17 import io.netty.util.concurrent.GenericFutureListener;
18 import java.util.ArrayDeque;
19 import java.util.Iterator;
20 import java.util.List;
21 import java.util.Queue;
22 import java.util.concurrent.locks.Lock;
23 import java.util.concurrent.locks.ReentrantLock;
24 import org.opendaylight.controller.netconf.api.NetconfDocumentedException;
25 import org.opendaylight.controller.netconf.api.NetconfMessage;
26 import org.opendaylight.controller.netconf.api.NetconfTerminationReason;
27 import org.opendaylight.controller.netconf.api.xml.XmlNetconfConstants;
28 import org.opendaylight.controller.netconf.client.NetconfClientDispatcher;
29 import org.opendaylight.controller.netconf.client.NetconfClientSession;
30 import org.opendaylight.controller.netconf.client.NetconfClientSessionListener;
31 import org.opendaylight.controller.netconf.client.conf.NetconfClientConfiguration;
32 import org.opendaylight.controller.netconf.client.conf.NetconfReconnectingClientConfiguration;
33 import org.opendaylight.controller.netconf.util.xml.XmlElement;
34 import org.opendaylight.controller.netconf.util.xml.XmlUtil;
35 import org.opendaylight.controller.sal.connect.api.RemoteDevice;
36 import org.opendaylight.controller.sal.connect.api.RemoteDeviceCommunicator;
37 import org.opendaylight.controller.sal.connect.netconf.util.NetconfMessageTransformUtil;
38 import org.opendaylight.controller.sal.connect.util.RemoteDeviceId;
39 import org.opendaylight.yangtools.yang.common.QName;
40 import org.opendaylight.yangtools.yang.common.RpcError;
41 import org.opendaylight.yangtools.yang.common.RpcResult;
42 import org.opendaylight.yangtools.yang.common.RpcResultBuilder;
43 import org.slf4j.Logger;
44 import org.slf4j.LoggerFactory;
45
46 public class NetconfDeviceCommunicator implements NetconfClientSessionListener, RemoteDeviceCommunicator<NetconfMessage> {
47
48     private static final Logger logger = LoggerFactory.getLogger(NetconfDeviceCommunicator.class);
49
50     private final RemoteDevice<NetconfSessionPreferences, NetconfMessage, NetconfDeviceCommunicator> remoteDevice;
51     private final Optional<NetconfSessionPreferences> overrideNetconfCapabilities;
52     private final RemoteDeviceId id;
53     private final Lock sessionLock = new ReentrantLock();
54
55     // TODO implement concurrent message limit
56     private final Queue<Request> requests = new ArrayDeque<>();
57     private NetconfClientSession session;
58     private Future<?> initFuture;
59
60     public NetconfDeviceCommunicator(final RemoteDeviceId id, final RemoteDevice<NetconfSessionPreferences, NetconfMessage, NetconfDeviceCommunicator> remoteDevice,
61             final NetconfSessionPreferences NetconfSessionPreferences) {
62         this(id, remoteDevice, Optional.of(NetconfSessionPreferences));
63     }
64
65     public NetconfDeviceCommunicator(final RemoteDeviceId id,
66                                      final RemoteDevice<NetconfSessionPreferences, NetconfMessage, NetconfDeviceCommunicator> remoteDevice) {
67         this(id, remoteDevice, Optional.<NetconfSessionPreferences>absent());
68     }
69
70     private NetconfDeviceCommunicator(final RemoteDeviceId id, final RemoteDevice<NetconfSessionPreferences, NetconfMessage, NetconfDeviceCommunicator> remoteDevice,
71             final Optional<NetconfSessionPreferences> overrideNetconfCapabilities) {
72         this.id = id;
73         this.remoteDevice = remoteDevice;
74         this.overrideNetconfCapabilities = overrideNetconfCapabilities;
75     }
76
77     @Override
78     public void onSessionUp(final NetconfClientSession session) {
79         sessionLock.lock();
80         try {
81             logger.debug("{}: Session established", id);
82             this.session = session;
83
84             NetconfSessionPreferences netconfSessionPreferences =
85                                              NetconfSessionPreferences.fromNetconfSession(session);
86             logger.trace("{}: Session advertised capabilities: {}", id, netconfSessionPreferences);
87
88             if(overrideNetconfCapabilities.isPresent()) {
89                 netconfSessionPreferences = netconfSessionPreferences.replaceModuleCaps(overrideNetconfCapabilities.get());
90                 logger.debug("{}: Session capabilities overridden, capabilities that will be used: {}", id, netconfSessionPreferences);
91             }
92
93             remoteDevice.onRemoteSessionUp(netconfSessionPreferences, this);
94         }
95         finally {
96             sessionLock.unlock();
97         }
98     }
99
100     public void initializeRemoteConnection(final NetconfClientDispatcher dispatcher, final NetconfClientConfiguration config) {
101         // TODO 2313 extract listener from configuration
102         if(config instanceof NetconfReconnectingClientConfiguration) {
103             initFuture = dispatcher.createReconnectingClient((NetconfReconnectingClientConfiguration) config);
104         } else {
105             initFuture = dispatcher.createClient(config);
106         }
107
108
109         initFuture.addListener(new GenericFutureListener<Future<Object>>(){
110
111             @Override
112             public void operationComplete(Future<Object> future) throws Exception {
113                 if (!future.isSuccess()) {
114                     logger.debug("{}: Connection failed", id, future.cause());
115                     NetconfDeviceCommunicator.this.remoteDevice.onRemoteSessionFailed(future.cause());
116                 }
117             }
118         });
119
120     }
121
122     public void disconnect() {
123         if(session != null) {
124             session.close();
125         }
126     }
127
128     private void tearDown( String reason ) {
129         List<UncancellableFuture<RpcResult<NetconfMessage>>> futuresToCancel = Lists.newArrayList();
130         sessionLock.lock();
131         try {
132             if( session != null ) {
133                 session = null;
134
135                 /*
136                  * Walk all requests, check if they have been executing
137                  * or cancelled and remove them from the queue.
138                  */
139                 final Iterator<Request> it = requests.iterator();
140                 while (it.hasNext()) {
141                     final Request r = it.next();
142                     if (r.future.isUncancellable()) {
143                         futuresToCancel.add( r.future );
144                         it.remove();
145                     } else if (r.future.isCancelled()) {
146                         // This just does some house-cleaning
147                         it.remove();
148                     }
149                 }
150
151                 remoteDevice.onRemoteSessionDown();
152             }
153         }
154         finally {
155             sessionLock.unlock();
156         }
157
158         // Notify pending request futures outside of the sessionLock to avoid unnecessarily
159         // blocking the caller.
160         for( UncancellableFuture<RpcResult<NetconfMessage>> future: futuresToCancel ) {
161             if( Strings.isNullOrEmpty( reason ) ) {
162                 future.set( createSessionDownRpcResult() );
163             } else {
164                 future.set( createErrorRpcResult( RpcError.ErrorType.TRANSPORT, reason ) );
165             }
166         }
167     }
168
169     private RpcResult<NetconfMessage> createSessionDownRpcResult() {
170         return createErrorRpcResult( RpcError.ErrorType.TRANSPORT,
171                              String.format( "The netconf session to %1$s is disconnected", id.getName() ) );
172     }
173
174     private RpcResult<NetconfMessage> createErrorRpcResult( RpcError.ErrorType errorType, String message ) {
175         return RpcResultBuilder.<NetconfMessage>failed()
176                 .withError(errorType, NetconfDocumentedException.ErrorTag.operation_failed.getTagValue(), message).build();
177     }
178
179     @Override
180     public void onSessionDown(final NetconfClientSession session, final Exception e) {
181         logger.warn("{}: Session went down", id, e);
182         tearDown( null );
183     }
184
185     @Override
186     public void onSessionTerminated(final NetconfClientSession session, final NetconfTerminationReason reason) {
187         logger.warn("{}: Session terminated {}", id, reason);
188         tearDown( reason.getErrorMessage() );
189     }
190
191     @Override
192     public void close() {
193         // Cancel reconnect if in progress
194         if(initFuture != null) {
195             initFuture.cancel(false);
196         }
197         // Disconnect from device
198         if(session != null) {
199             session.close();
200         }
201
202         tearDown(id + ": Netconf session closed");
203     }
204
205     @Override
206     public void onMessage(final NetconfClientSession session, final NetconfMessage message) {
207         /*
208          * Dispatch between notifications and messages. Messages need to be processed
209          * with lock held, notifications do not.
210          */
211         if (isNotification(message)) {
212             processNotification(message);
213         } else {
214             processMessage(message);
215         }
216     }
217
218     private void processMessage(final NetconfMessage message) {
219         Request request = null;
220         sessionLock.lock();
221
222         try {
223             request = requests.peek();
224             if (request != null && request.future.isUncancellable()) {
225                 requests.poll();
226             } else {
227                 request = null;
228                 logger.warn("{}: Ignoring unsolicited message {}", id, msgToS(message));
229             }
230         }
231         finally {
232             sessionLock.unlock();
233         }
234
235         if( request != null ) {
236
237             logger.debug("{}: Message received {}", id, message);
238
239             if(logger.isTraceEnabled()) {
240                 logger.trace( "{}: Matched request: {} to response: {}", id, msgToS( request.request ), msgToS( message ) );
241             }
242
243             try {
244                 NetconfMessageTransformUtil.checkValidReply( request.request, message );
245             } catch (final NetconfDocumentedException e) {
246                 logger.warn( "{}: Invalid request-reply match, reply message contains different message-id, request: {}, response: {}",
247                              id, msgToS( request.request ), msgToS( message ), e );
248
249                 request.future.set( RpcResultBuilder.<NetconfMessage>failed()
250                         .withRpcError( NetconfMessageTransformUtil.toRpcError( e ) ).build() );
251                 return;
252             }
253
254             try {
255                 NetconfMessageTransformUtil.checkSuccessReply(message);
256             } catch(final NetconfDocumentedException e) {
257                 logger.warn( "{}: Error reply from remote device, request: {}, response: {}", id,
258                              msgToS( request.request ), msgToS( message ), e );
259
260                 request.future.set( RpcResultBuilder.<NetconfMessage>failed()
261                         .withRpcError( NetconfMessageTransformUtil.toRpcError( e ) ).build() );
262                 return;
263             }
264
265             request.future.set( RpcResultBuilder.success( message ).build() );
266         }
267     }
268
269     private static String msgToS(final NetconfMessage msg) {
270         return XmlUtil.toString(msg.getDocument());
271     }
272
273     @Override
274     public ListenableFuture<RpcResult<NetconfMessage>> sendRequest(final NetconfMessage message, final QName rpc) {
275         sessionLock.lock();
276         try {
277             return sendRequestWithLock( message, rpc );
278         } finally {
279             sessionLock.unlock();
280         }
281     }
282
283     private ListenableFuture<RpcResult<NetconfMessage>> sendRequestWithLock(
284                                                final NetconfMessage message, final QName rpc) {
285         if(logger.isTraceEnabled()) {
286             logger.trace("{}: Sending message {}", id, msgToS(message));
287         }
288
289         if (session == null) {
290             logger.warn("{}: Session is disconnected, failing RPC request {}", id, message);
291             return Futures.immediateFuture( createSessionDownRpcResult() );
292         }
293
294         final Request req = new Request( new UncancellableFuture<RpcResult<NetconfMessage>>(true),
295                                          message );
296         requests.add(req);
297
298         session.sendMessage(req.request).addListener(new FutureListener<Void>() {
299             @Override
300             public void operationComplete(final Future<Void> future) throws Exception {
301                 if( !future.isSuccess() ) {
302                     // We expect that a session down will occur at this point
303                     logger.debug( "{}: Failed to send request {}", id,
304                                   XmlUtil.toString(req.request.getDocument()), future.cause() );
305
306                     if( future.cause() != null ) {
307                         req.future.set( createErrorRpcResult( RpcError.ErrorType.TRANSPORT,
308                                                               future.cause().getLocalizedMessage() ) );
309                     } else {
310                         req.future.set( createSessionDownRpcResult() ); // assume session is down
311                     }
312                     req.future.setException( future.cause() );
313                 }
314                 else {
315                     logger.trace( "Finished sending request {}", req.request );
316                 }
317             }
318         });
319
320         return req.future;
321     }
322
323     private void processNotification(final NetconfMessage notification) {
324         logger.debug("{}: Notification received: {}", id, notification);
325
326         if(logger.isTraceEnabled()) {
327             logger.trace("{}: Notification received: {}", id, msgToS(notification));
328         }
329
330         remoteDevice.onNotification(notification);
331     }
332
333     private static boolean isNotification(final NetconfMessage message) {
334         final XmlElement xmle = XmlElement.fromDomDocument(message.getDocument());
335         return XmlNetconfConstants.NOTIFICATION_ELEMENT_NAME.equals(xmle.getName()) ;
336     }
337
338     private static final class Request {
339         final UncancellableFuture<RpcResult<NetconfMessage>> future;
340         final NetconfMessage request;
341
342         private Request(final UncancellableFuture<RpcResult<NetconfMessage>> future,
343                         final NetconfMessage request) {
344             this.future = future;
345             this.request = request;
346         }
347     }
348 }

©2013 OpenDaylight, A Linux Foundation Collaborative Project. All Rights Reserved.
OpenDaylight is a registered trademark of The OpenDaylight Project, Inc.
Linux Foundation and OpenDaylight are registered trademarks of the Linux Foundation.
Linux is a registered trademark of Linus Torvalds.