restconf/restconf-nb/src/main/java/org/opendaylight/restconf/nb/rfc8040/streams/WebSocketInitializer.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 com.google.common.annotations.VisibleForTesting;
  11 import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
  12 import java.io.Serial;
  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.ListenersBroker;
  24 import org.slf4j.Logger;
  25 import org.slf4j.LoggerFactory;
  26
  27 /**
  28  * Web-socket servlet listening on ws or wss schemas for created data-change-event or notification streams.
  29  */
  30 @Singleton
  31 public final class WebSocketInitializer extends WebSocketServlet {
  32     @Serial
  33     private static final long serialVersionUID = 1L;
  34
  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;
  41
  42     /**
  43      * Creation of the web-socket initializer.
  44      *
  45      * @param scheduledThreadPool    ODL thread pool used for fetching of scheduled executors.
  46      * @param configuration          Web-socket configuration holder.
  47      */
  48     @Inject
  49     public WebSocketInitializer(final ScheduledThreadPool scheduledThreadPool,
  50             final StreamsConfiguration configuration) {
  51         executorService = scheduledThreadPool.getExecutor();
  52         maximumFragmentLength = configuration.maximumFragmentLength();
  53         heartbeatInterval = configuration.heartbeatInterval();
  54         idleTimeoutMillis = configuration.idleTimeout();
  55     }
  56
  57     /**
  58      * Configuration of the web-socket factory - idle timeout and specified factory object.
  59      *
  60      * @param factory Configurable web-socket factory.
  61      */
  62     @Override
  63     public void configure(final WebSocketServletFactory factory) {
  64         factory.getPolicy().setIdleTimeout(idleTimeoutMillis);
  65         factory.setCreator(new WebSocketFactory(executorService, maximumFragmentLength, heartbeatInterval));
  66     }
  67
  68     /**
  69      * Factory that is used for creation of new web-sockets based on HTTP/HTTPS upgrade request.
  70      */
  71     @VisibleForTesting
  72     static final class WebSocketFactory implements WebSocketCreator {
  73         private static final Logger LOG = LoggerFactory.getLogger(WebSocketFactory.class);
  74
  75         private final ScheduledExecutorService executorService;
  76         // FIXME: inject this reference
  77         private final ListenersBroker listenersBroker = ListenersBroker.getInstance();
  78         private final int maximumFragmentLength;
  79         private final int heartbeatInterval;
  80
  81         /**
  82          * Creation of the web-socket factory.
  83          *
  84          * @param executorService       Executor for creation of threads for controlling of web-socket sessions.
  85          * @param maximumFragmentLength Maximum web-socket fragment length in number of Unicode code units (characters)
  86          *                              (exceeded message length leads to fragmentation of messages).
  87          * @param heartbeatInterval     Interval in milliseconds between sending of ping control frames.
  88          */
  89         WebSocketFactory(final ScheduledExecutorService executorService, final int maximumFragmentLength,
  90                 final int heartbeatInterval) {
  91             this.executorService = executorService;
  92             this.maximumFragmentLength = maximumFragmentLength;
  93             this.heartbeatInterval = heartbeatInterval;
  94         }
  95
  96         /**
  97          * Creation of the new web-socket based on input HTTP/HTTPS upgrade request. Web-socket is created only if the
  98          * data listener for input URI can be found (results in status code 101); otherwise status code 404 is set
  99          * in upgrade response.
 100          *
 101          * @param servletUpgradeRequest  Upgrade request.
 102          * @param servletUpgradeResponse Upgrade response.
 103          * @return Created web-socket instance or {@code null} if the web-socket cannot be created.
 104          */
 105         @Override
 106         public Object createWebSocket(final ServletUpgradeRequest servletUpgradeRequest,
 107                 final ServletUpgradeResponse servletUpgradeResponse) {
 108             final var streamName = ListenersBroker.createStreamNameFromUri(
 109                 servletUpgradeRequest.getRequestURI().getRawPath());
 110
 111             final var listener = listenersBroker.listenerFor(streamName);
 112             if (listener == null) {
 113                 LOG.debug("Listener for stream with name {} was not found.", streamName);
 114                 servletUpgradeResponse.setSuccess(false);
 115                 servletUpgradeResponse.setStatusCode(HttpServletResponse.SC_NOT_FOUND);
 116                 return null;
 117             }
 118
 119             LOG.debug("Listener for stream with name {} has been found, web-socket session handler will be created",
 120                 streamName);
 121             servletUpgradeResponse.setSuccess(true);
 122             servletUpgradeResponse.setStatusCode(HttpServletResponse.SC_SWITCHING_PROTOCOLS);
 123             // note: every web-socket manages PING process individually because this approach scales better than
 124             //       sending of PING frames at once over all web-socket sessions
 125             return new WebSocketSessionHandler(executorService, listener, maximumFragmentLength, heartbeatInterval);
 126         }
 127     }
 128 }