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