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.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;
  27
  28 /**
  29  * Web-socket servlet listening on ws or wss schemas for created data-change-event or notification streams.
  30  */
  31 @Singleton
  32 public final class WebSocketInitializer extends WebSocketServlet {
  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, final Configuration configuration) {
  50         this.executorService = scheduledThreadPool.getExecutor();
  51         this.maximumFragmentLength = configuration.getMaximumFragmentLength();
  52         this.heartbeatInterval = configuration.getHeartbeatInterval();
  53         this.idleTimeoutMillis = configuration.getIdleTimeout();
  54     }
  55
  56     /**
  57      * Configuration of the web-socket factory - idle timeout and specified factory object.
  58      *
  59      * @param factory Configurable web-socket factory.
  60      */
  61     @Override
  62     public void configure(final WebSocketServletFactory factory) {
  63         factory.getPolicy().setIdleTimeout(idleTimeoutMillis);
  64         factory.setCreator(new WebSocketFactory(executorService, maximumFragmentLength, heartbeatInterval));
  65     }
  66
  67     /**
  68      * Factory that is used for creation of new web-sockets based on HTTP/HTTPS upgrade request.
  69      */
  70     @VisibleForTesting
  71     static final class WebSocketFactory implements WebSocketCreator {
  72         private static final Logger LOG = LoggerFactory.getLogger(WebSocketFactory.class);
  73
  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;
  79
  80         /**
  81          * Creation of the web-socket factory.
  82          *
  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.
  87          */
  88         WebSocketFactory(final ScheduledExecutorService executorService, final int maximumFragmentLength,
  89                 final int heartbeatInterval) {
  90             this.executorService = executorService;
  91             this.maximumFragmentLength = maximumFragmentLength;
  92             this.heartbeatInterval = heartbeatInterval;
  93         }
  94
  95         /**
  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.
  99          *
 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.
 103          */
 104         @Override
 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);
 109
 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",
 113                         streamName);
 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,
 119                         heartbeatInterval);
 120             } else {
 121                 LOG.debug("Listener for stream with name {} was not found.", streamName);
 122                 servletUpgradeResponse.setSuccess(false);
 123                 servletUpgradeResponse.setStatusCode(HttpServletResponse.SC_NOT_FOUND);
 124                 return null;
 125             }
 126         }
 127     }
 128 }