Expose streams with all supported encodings
[netconf.git] / restconf / restconf-nb / src / main / java / org / opendaylight / restconf / nb / rfc8040 / streams / WebSocketSessionHandler.java
1 /*
2  * Copyright © 2019 FRINX s.r.o. 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.restconf.nb.rfc8040.streams;
9
10 import static java.util.Objects.requireNonNull;
11
12 import com.google.common.base.Strings;
13 import java.io.IOException;
14 import java.io.UnsupportedEncodingException;
15 import java.nio.ByteBuffer;
16 import java.nio.charset.Charset;
17 import java.util.ArrayList;
18 import java.util.List;
19 import java.util.Objects;
20 import java.util.concurrent.ScheduledExecutorService;
21 import java.util.concurrent.ScheduledFuture;
22 import java.util.concurrent.TimeUnit;
23 import java.util.concurrent.TimeoutException;
24 import javax.xml.xpath.XPathExpressionException;
25 import org.eclipse.jdt.annotation.Nullable;
26 import org.eclipse.jetty.websocket.api.CloseException;
27 import org.eclipse.jetty.websocket.api.RemoteEndpoint;
28 import org.eclipse.jetty.websocket.api.Session;
29 import org.eclipse.jetty.websocket.api.annotations.OnWebSocketClose;
30 import org.eclipse.jetty.websocket.api.annotations.OnWebSocketConnect;
31 import org.eclipse.jetty.websocket.api.annotations.OnWebSocketError;
32 import org.eclipse.jetty.websocket.api.annotations.WebSocket;
33 import org.opendaylight.restconf.nb.rfc8040.ReceiveEventsParams;
34 import org.opendaylight.restconf.nb.rfc8040.streams.RestconfStream.EncodingName;
35 import org.opendaylight.yangtools.concepts.Registration;
36 import org.slf4j.Logger;
37 import org.slf4j.LoggerFactory;
38
39 /**
40  * Web-socket session handler that is responsible for controlling of session, managing subscription
41  * to data-change-event or notification listener, and sending of data over established web-socket session.
42  */
43 @WebSocket
44 public final class WebSocketSessionHandler implements StreamSessionHandler {
45     private static final Logger LOG = LoggerFactory.getLogger(WebSocketSessionHandler.class);
46     private static final byte[] PING_PAYLOAD = "ping".getBytes(Charset.defaultCharset());
47
48     private final ScheduledExecutorService executorService;
49     private final RestconfStream<?> stream;
50     private final EncodingName encodingName;
51     private final ReceiveEventsParams params;
52     private final int maximumFragmentLength;
53     private final int heartbeatInterval;
54
55     private Session session;
56     private Registration subscriber;
57     private ScheduledFuture<?> pingProcess;
58
59     /**
60      * Creation of the new web-socket session handler.
61      *
62      * @param executorService       Executor that is used for periodical sending of web-socket ping messages to keep
63      *                              session up even if the notifications doesn't flow from server to clients or clients
64      *                              don't implement ping-pong service.
65      * @param stream                YANG notification or data-change event listener to which client on this web-socket
66      *                              session subscribes to.
67      * @param maximumFragmentLength Maximum fragment length in number of Unicode code units (characters).
68      *                              If this parameter is set to 0, the maximum fragment length is disabled and messages
69      *                              up to 64 KB can be sent in TCP segment (exceeded notification length ends in error).
70      *                              If the parameter is set to non-zero positive value, messages longer than this
71      *                              parameter are fragmented into multiple web-socket messages sent in one transaction.
72      * @param heartbeatInterval     Interval in milliseconds of sending of ping control frames to remote endpoint
73      *                              to keep session up. Ping control frames are disabled if this parameter is set to 0.
74      */
75     WebSocketSessionHandler(final ScheduledExecutorService executorService, final RestconfStream<?> stream,
76             final EncodingName encodingName, final @Nullable ReceiveEventsParams params,
77             final int maximumFragmentLength, final int heartbeatInterval) {
78         this.executorService = requireNonNull(executorService);
79         this.stream = requireNonNull(stream);
80         this.encodingName = requireNonNull(encodingName);
81         // FIXME: NETCONF-1102: require params
82         this.params = params;
83         this.maximumFragmentLength = maximumFragmentLength;
84         this.heartbeatInterval = heartbeatInterval;
85     }
86
87     /**
88      * Handling of the web-socket connected event (web-socket session between local server and remote endpoint has been
89      * established). Web-socket session handler is registered at data-change-event / YANG notification listener and
90      * the heartbeat ping process is started if it is enabled.
91      *
92      * @param webSocketSession Created web-socket session.
93      * @see OnWebSocketConnect More information about invocation of this method and parameters.
94      */
95     @OnWebSocketConnect
96     public synchronized void onWebSocketConnected(final Session webSocketSession) {
97         if (session == null || !session.isOpen()) {
98             session = webSocketSession;
99             try {
100                 subscriber = stream.addSubscriber(this, encodingName, params);
101             } catch (IllegalArgumentException | XPathExpressionException | UnsupportedEncodingException e) {
102                 LOG.info("Closing web-socket session {}", webSocketSession, e);
103                 webSocketSession.close(404, "Unsupported encoding " + encodingName);
104                 session = null;
105                 return;
106             }
107
108             LOG.debug("A new web-socket session {} has been successfully registered.", webSocketSession);
109             if (heartbeatInterval != 0) {
110                 // sending of PING frame can be long if there is an error on web-socket - from this reason
111                 // the fixed-rate should not be used
112                 pingProcess = executorService.scheduleWithFixedDelay(this::sendPingMessage, heartbeatInterval,
113                         heartbeatInterval, TimeUnit.MILLISECONDS);
114             }
115         }
116     }
117
118     /**
119      * Handling of web-socket session closed event (timeout, error, or both parties closed session). Removal
120      * of subscription at listener and stopping of the ping process.
121      *
122      * @param statusCode Web-socket status code.
123      * @param reason     Reason, why the web-socket is closed (for example, reached timeout).
124      * @see OnWebSocketClose More information about invocation of this method and parameters.
125      */
126     @OnWebSocketClose
127     public synchronized void onWebSocketClosed(final int statusCode, final String reason) {
128         // note: there is not guarantee that Session.isOpen() returns true - it is better to not check it here
129         // using 'session != null && session.isOpen()'
130         if (session != null) {
131             LOG.debug("Web-socket session has been closed with status code {} and reason message: {}.", statusCode,
132                 reason);
133             if (subscriber != null) {
134                 subscriber.close();
135                 subscriber = null;
136             }
137             stopPingProcess();
138         }
139     }
140
141     /**
142      * Handling of error in web-socket implementation. Subscription at listener is removed, open session is closed
143      * and ping process is stopped.
144      *
145      * @param error Error details.
146      * @see OnWebSocketError More information about invocation of this method and parameters.
147      */
148     @OnWebSocketError
149     public synchronized void onWebSocketError(final Throwable error) {
150         if (error instanceof CloseException && error.getCause() instanceof TimeoutException timeout) {
151             // A timeout is expected, do not log the complete stack trace
152             LOG.info("Web-socket closed by timeout: {}", timeout.getMessage());
153         } else {
154             LOG.warn("An error occurred on web-socket: ", error);
155         }
156         if (session != null) {
157             LOG.info("Trying to close web-socket session {} gracefully after error.", session);
158             if (subscriber != null) {
159                 subscriber.close();
160                 subscriber = null;
161             }
162             if (session.isOpen()) {
163                 session.close();
164             }
165             stopPingProcess();
166         }
167     }
168
169     private void stopPingProcess() {
170         if (pingProcess != null && !pingProcess.isDone() && !pingProcess.isCancelled()) {
171             pingProcess.cancel(true);
172         }
173     }
174
175     @Override
176     public synchronized void endOfStream() {
177         if (session != null && session.isOpen()) {
178             session.close();
179         }
180         stopPingProcess();
181     }
182
183     /**
184      * Sensing of string message to remote endpoint of {@link org.eclipse.jetty.websocket.api.Session}. If the maximum
185      * fragment length is set to non-zero positive value and input message exceeds this value, message is fragmented
186      * to multiple message fragments which are send individually but still in one web-socket transaction.
187      *
188      * @param message Message data to be send over web-socket session.
189      */
190     @Override
191     public synchronized void sendDataMessage(final String message) {
192         if (Strings.isNullOrEmpty(message)) {
193             // FIXME: should this be tolerated?
194             return;
195         }
196
197         if (session != null && session.isOpen()) {
198             final var remoteEndpoint = session.getRemote();
199             if (maximumFragmentLength == 0 || message.length() <= maximumFragmentLength) {
200                 sendDataMessage(message, remoteEndpoint);
201             } else {
202                 sendFragmentedMessage(splitMessageToFragments(message, maximumFragmentLength), remoteEndpoint);
203             }
204         } else {
205             LOG.trace("Message with body '{}' is not sent because underlay web-socket session is not open.", message);
206         }
207     }
208
209     private void sendDataMessage(final String message, final RemoteEndpoint remoteEndpoint) {
210         try {
211             remoteEndpoint.sendString(message);
212             LOG.trace("Message with body '{}' has been successfully sent to remote endpoint {}.", message,
213                 remoteEndpoint);
214         } catch (IOException e) {
215             LOG.warn("Cannot send message over web-socket session {}.", session, e);
216         }
217     }
218
219     private void sendFragmentedMessage(final List<String> orderedFragments, final RemoteEndpoint remoteEndpoint) {
220         for (int i = 0; i < orderedFragments.size(); i++) {
221             final String fragment = orderedFragments.get(i);
222             final boolean last = i == orderedFragments.size() - 1;
223
224             try {
225                 remoteEndpoint.sendPartialString(fragment, last);
226             } catch (IOException e) {
227                 LOG.warn("Cannot send message fragment number {} over web-socket session {}. All other fragments of "
228                     + " the message are dropped too.", i, session, e);
229                 return;
230             }
231             LOG.trace("Message fragment number {} with body '{}' has been successfully sent to remote endpoint {}.", i,
232                 fragment, remoteEndpoint);
233         }
234     }
235
236     private synchronized void sendPingMessage() {
237         try {
238             Objects.requireNonNull(session).getRemote().sendPing(ByteBuffer.wrap(PING_PAYLOAD));
239         } catch (IOException e) {
240             LOG.warn("Cannot send ping message over web-socket session {}.", session, e);
241         }
242     }
243
244     private static List<String> splitMessageToFragments(final String inputMessage, final int maximumFragmentLength) {
245         final var parts = new ArrayList<String>();
246         int length = inputMessage.length();
247         for (int i = 0; i < length; i += maximumFragmentLength) {
248             parts.add(inputMessage.substring(i, Math.min(length, i + maximumFragmentLength)));
249         }
250         return parts;
251     }
252 }