2 * Copyright © 2019 FRINX s.r.o. All rights reserved.
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
8 package org.opendaylight.restconf.nb.rfc8040.streams;
10 import com.google.common.annotations.VisibleForTesting;
11 import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
12 import java.util.Optional;
13 import java.util.concurrent.ScheduledExecutorService;
14 import javax.inject.Inject;
15 import javax.inject.Singleton;
16 import javax.servlet.http.HttpServletResponse;
17 import org.eclipse.jetty.websocket.servlet.ServletUpgradeRequest;
18 import org.eclipse.jetty.websocket.servlet.ServletUpgradeResponse;
19 import org.eclipse.jetty.websocket.servlet.WebSocketCreator;
20 import org.eclipse.jetty.websocket.servlet.WebSocketServlet;
21 import org.eclipse.jetty.websocket.servlet.WebSocketServletFactory;
22 import org.opendaylight.controller.config.threadpool.ScheduledThreadPool;
23 import org.opendaylight.restconf.nb.rfc8040.streams.listeners.BaseListenerInterface;
24 import org.opendaylight.restconf.nb.rfc8040.streams.listeners.ListenersBroker;
25 import org.slf4j.Logger;
26 import org.slf4j.LoggerFactory;
29 * Web-socket servlet listening on ws or wss schemas for created data-change-event or notification streams.
32 public final class WebSocketInitializer extends WebSocketServlet {
33 private static final long serialVersionUID = 1L;
35 @SuppressFBWarnings(value = "SE_BAD_FIELD",
36 justification = "Servlet/WebSocket bridge, we need this service for heartbeats")
37 private final ScheduledExecutorService executorService;
38 private final int maximumFragmentLength;
39 private final int heartbeatInterval;
40 private final int idleTimeoutMillis;
43 * Creation of the web-socket initializer.
45 * @param scheduledThreadPool ODL thread pool used for fetching of scheduled executors.
46 * @param configuration Web-socket configuration holder.
49 public WebSocketInitializer(final ScheduledThreadPool scheduledThreadPool, final Configuration configuration) {
50 this.executorService = scheduledThreadPool.getExecutor();
51 this.maximumFragmentLength = configuration.getMaximumFragmentLength();
52 this.heartbeatInterval = configuration.getHeartbeatInterval();
53 this.idleTimeoutMillis = configuration.getIdleTimeout();
57 * Configuration of the web-socket factory - idle timeout and specified factory object.
59 * @param factory Configurable web-socket factory.
62 public void configure(final WebSocketServletFactory factory) {
63 factory.getPolicy().setIdleTimeout(idleTimeoutMillis);
64 factory.setCreator(new WebSocketFactory(executorService, maximumFragmentLength, heartbeatInterval));
68 * Factory that is used for creation of new web-sockets based on HTTP/HTTPS upgrade request.
71 static final class WebSocketFactory implements WebSocketCreator {
72 private static final Logger LOG = LoggerFactory.getLogger(WebSocketFactory.class);
74 private final ScheduledExecutorService executorService;
75 // FIXME: inject this reference
76 private final ListenersBroker listenersBroker = ListenersBroker.getInstance();
77 private final int maximumFragmentLength;
78 private final int heartbeatInterval;
81 * Creation of the web-socket factory.
83 * @param executorService Executor for creation of threads for controlling of web-socket sessions.
84 * @param maximumFragmentLength Maximum web-socket fragment length in number of Unicode code units (characters)
85 * (exceeded message length leads to fragmentation of messages).
86 * @param heartbeatInterval Interval in milliseconds between sending of ping control frames.
88 WebSocketFactory(final ScheduledExecutorService executorService, final int maximumFragmentLength,
89 final int heartbeatInterval) {
90 this.executorService = executorService;
91 this.maximumFragmentLength = maximumFragmentLength;
92 this.heartbeatInterval = heartbeatInterval;
96 * Creation of the new web-socket based on input HTTP/HTTPS upgrade request. Web-socket is created only if the
97 * data listener for input URI can be found (results in status code 101); otherwise status code 404 is set
98 * in upgrade response.
100 * @param servletUpgradeRequest Upgrade request.
101 * @param servletUpgradeResponse Upgrade response.
102 * @return Created web-socket instance or {@code null} if the web-socket cannot be created.
105 public Object createWebSocket(final ServletUpgradeRequest servletUpgradeRequest,
106 final ServletUpgradeResponse servletUpgradeResponse) {
107 final String requestUri = servletUpgradeRequest.getRequestURI().getRawPath();
108 final String streamName = ListenersBroker.createStreamNameFromUri(requestUri);
110 final Optional<BaseListenerInterface> listener = listenersBroker.getListenerFor(streamName);
111 if (listener.isPresent()) {
112 LOG.debug("Listener for stream with name {} has been found, web-socket session handler will be created",
114 servletUpgradeResponse.setSuccess(true);
115 servletUpgradeResponse.setStatusCode(HttpServletResponse.SC_SWITCHING_PROTOCOLS);
116 // note: every web-socket manages PING process individually because this approach scales better than
117 // sending of PING frames at once over all web-socket sessions
118 return new WebSocketSessionHandler(executorService, listener.get(), maximumFragmentLength,
121 LOG.debug("Listener for stream with name {} was not found.", streamName);
122 servletUpgradeResponse.setSuccess(false);
123 servletUpgradeResponse.setStatusCode(HttpServletResponse.SC_NOT_FOUND);