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

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