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

©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.