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> 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> 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> remoteDevice) {
67         this(id, remoteDevice, Optional.<NetconfSessionPreferences>absent());
68     }
69
70     private NetconfDeviceCommunicator(final RemoteDeviceId id, final RemoteDevice<NetconfSessionPreferences, NetconfMessage> 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 dispatch,
101                                            final NetconfClientConfiguration config) {
102         if(config instanceof NetconfReconnectingClientConfiguration) {
103             initFuture = dispatch.createReconnectingClient((NetconfReconnectingClientConfiguration) config);
104         } else {
105             initFuture = dispatch.createClient(config);
106         }
107
108         initFuture.addListener(new GenericFutureListener<Future<Object>>(){
109
110             @Override
111             public void operationComplete(Future<Object> future) throws Exception {
112                 if (!future.isSuccess()) {
113                     logger.debug("{}: Connection failed", id, future.cause());
114                     NetconfDeviceCommunicator.this.remoteDevice.onRemoteSessionFailed(future.cause());
115                 }
116             }
117         });
118     }
119
120     private void tearDown( String reason ) {
121         List<UncancellableFuture<RpcResult<NetconfMessage>>> futuresToCancel = Lists.newArrayList();
122         sessionLock.lock();
123         try {
124             if( session != null ) {
125                 session = null;
126
127                 /*
128                  * Walk all requests, check if they have been executing
129                  * or cancelled and remove them from the queue.
130                  */
131                 final Iterator<Request> it = requests.iterator();
132                 while (it.hasNext()) {
133                     final Request r = it.next();
134                     if (r.future.isUncancellable()) {
135                         futuresToCancel.add( r.future );
136                         it.remove();
137                     } else if (r.future.isCancelled()) {
138                         // This just does some house-cleaning
139                         it.remove();
140                     }
141                 }
142
143                 remoteDevice.onRemoteSessionDown();
144             }
145         }
146         finally {
147             sessionLock.unlock();
148         }
149
150         // Notify pending request futures outside of the sessionLock to avoid unnecessarily
151         // blocking the caller.
152         for( UncancellableFuture<RpcResult<NetconfMessage>> future: futuresToCancel ) {
153             if( Strings.isNullOrEmpty( reason ) ) {
154                 future.set( createSessionDownRpcResult() );
155             } else {
156                 future.set( createErrorRpcResult( RpcError.ErrorType.TRANSPORT, reason ) );
157             }
158         }
159     }
160
161     private RpcResult<NetconfMessage> createSessionDownRpcResult()
162     {
163         return createErrorRpcResult( RpcError.ErrorType.TRANSPORT,
164                              String.format( "The netconf session to %1$s is disconnected", id.getName() ) );
165     }
166
167     private RpcResult<NetconfMessage> createErrorRpcResult( RpcError.ErrorType errorType, String message )
168     {
169         return RpcResultBuilder.<NetconfMessage>failed()
170                 .withError( errorType, NetconfDocumentedException.ErrorTag.operation_failed.getTagValue(),
171                             message )
172                 .build();
173     }
174
175     @Override
176     public void onSessionDown(final NetconfClientSession session, final Exception e) {
177         logger.warn("{}: Session went down", id, e);
178         tearDown( null );
179     }
180
181     @Override
182     public void onSessionTerminated(final NetconfClientSession session, final NetconfTerminationReason reason) {
183         logger.warn("{}: Session terminated {}", id, reason);
184         tearDown( reason.getErrorMessage() );
185     }
186
187     @Override
188     public void close() {
189         // Cancel reconnect if in progress
190         if(initFuture != null) {
191             initFuture.cancel(false);
192         }
193         // Disconnect from device
194         if(session != null) {
195             session.close();
196         }
197         tearDown(id + ": Netconf session closed");
198     }
199
200     @Override
201     public void onMessage(final NetconfClientSession session, final NetconfMessage message) {
202         /*
203          * Dispatch between notifications and messages. Messages need to be processed
204          * with lock held, notifications do not.
205          */
206         if (isNotification(message)) {
207             processNotification(message);
208         } else {
209             processMessage(message);
210         }
211     }
212
213     private void processMessage(final NetconfMessage message) {
214         Request request = null;
215         sessionLock.lock();
216
217         try {
218             request = requests.peek();
219             if (request != null && request.future.isUncancellable()) {
220                 requests.poll();
221             } else {
222                 request = null;
223                 logger.warn("{}: Ignoring unsolicited message {}", id, msgToS(message));
224             }
225         }
226         finally {
227             sessionLock.unlock();
228         }
229
230         if( request != null ) {
231
232             logger.debug("{}: Message received {}", id, message);
233
234             if(logger.isTraceEnabled()) {
235                 logger.trace( "{}: Matched request: {} to response: {}", id,
236                               msgToS( request.request ), msgToS( message ) );
237             }
238
239             try {
240                 NetconfMessageTransformUtil.checkValidReply( request.request, message );
241             }
242             catch (final NetconfDocumentedException e) {
243                 logger.warn( "{}: Invalid request-reply match, reply message contains different message-id, request: {}, response: {}",
244                              id, msgToS( request.request ), msgToS( message ), e );
245
246                 request.future.set( RpcResultBuilder.<NetconfMessage>failed()
247                         .withRpcError( NetconfMessageTransformUtil.toRpcError( e ) ).build() );
248                 return;
249             }
250
251             try {
252                 NetconfMessageTransformUtil.checkSuccessReply(message);
253             }
254             catch(final NetconfDocumentedException e) {
255                 logger.warn( "{}: Error reply from remote device, request: {}, response: {}", id,
256                              msgToS( request.request ), msgToS( message ), e );
257
258                 request.future.set( RpcResultBuilder.<NetconfMessage>failed()
259                         .withRpcError( NetconfMessageTransformUtil.toRpcError( e ) ).build() );
260                 return;
261             }
262
263             request.future.set( RpcResultBuilder.success( message ).build() );
264         }
265     }
266
267     private static String msgToS(final NetconfMessage msg) {
268         return XmlUtil.toString(msg.getDocument());
269     }
270
271     @Override
272     public ListenableFuture<RpcResult<NetconfMessage>> sendRequest(
273                                                final NetconfMessage message, final QName rpc) {
274         sessionLock.lock();
275         try {
276             return sendRequestWithLock( message, rpc );
277         }
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.