Initialize notifications in CommunityRestConf

Initialize /notif endpoint and everything related to it inside CommunityRestConf.

Signed-off-by: tobias.pobocik <tobias.pobocik@pantheon.tech>

Author: Tobianas

lighty-examples/lighty-community-restconf-actions-app/src/main/java/io/lighty/examples/controllers/actions/Main.java

@@ -158,6 +158,7 @@ private void startLighty(final ControllerConfiguration controllerConfiguration,
                 .from(RestConfConfigUtils.getRestConfConfiguration(restconfConfiguration,
                     this.lightyController.getServices()))
                 .withLightyServer(jettyServerBuilder)
+                .withScheduledThreadPool(lightyController.getServices().getScheduledThreadPool())
                 .build();
 
         //3. start openApi and RestConf server

lighty-modules/lighty-restconf-nb-community/src/main/java/io/lighty/modules/northbound/restconf/community/impl/CommunityRestConf.java

@@ -7,6 +7,10 @@
  */
 package io.lighty.modules.northbound.restconf.community.impl;
 
+import static org.opendaylight.restconf.nb.rfc8040.rests.utils.RestconfStreamsConstants.DATA_SUBSCRIPTION;
+import static org.opendaylight.restconf.nb.rfc8040.rests.utils.RestconfStreamsConstants.NOTIFICATION_STREAM;
+import static org.opendaylight.restconf.nb.rfc8040.utils.RestconfConstants.NOTIF;
+
 import com.google.common.base.Stopwatch;
 import com.google.common.base.Throwables;
 import io.lighty.core.controller.api.AbstractLightyModule;
@@ -15,23 +19,30 @@
 import io.lighty.server.LightyServerBuilder;
 import java.net.InetAddress;
 import java.net.InetSocketAddress;
+import javax.servlet.ServletContext;
 import org.eclipse.jetty.server.Server;
 import org.eclipse.jetty.server.handler.ContextHandlerCollection;
 import org.eclipse.jetty.servlet.ServletContextHandler;
 import org.eclipse.jetty.servlet.ServletHolder;
+import org.eclipse.jetty.websocket.server.WebSocketServerFactory;
+import org.eclipse.jetty.websocket.servlet.WebSocketServletFactory;
 import org.glassfish.jersey.server.ResourceConfig;
 import org.glassfish.jersey.servlet.ServletContainer;
+import org.opendaylight.controller.config.threadpool.ScheduledThreadPool;
 import org.opendaylight.mdsal.dom.api.DOMActionService;
 import org.opendaylight.mdsal.dom.api.DOMDataBroker;
 import org.opendaylight.mdsal.dom.api.DOMMountPointService;
 import org.opendaylight.mdsal.dom.api.DOMNotificationService;
 import org.opendaylight.mdsal.dom.api.DOMRpcService;
 import org.opendaylight.mdsal.dom.api.DOMSchemaService;
+import org.opendaylight.restconf.nb.rfc8040.DataStreamApplication;
 import org.opendaylight.restconf.nb.rfc8040.RestconfApplication;
 import org.opendaylight.restconf.nb.rfc8040.databind.DatabindProvider;
 import org.opendaylight.restconf.nb.rfc8040.databind.mdsal.DOMDatabindProvider;
 import org.opendaylight.restconf.nb.rfc8040.handlers.SchemaContextHandler;
+import org.opendaylight.restconf.nb.rfc8040.rests.services.impl.RestconfDataStreamServiceImpl;
 import org.opendaylight.restconf.nb.rfc8040.streams.StreamsConfiguration;
+import org.opendaylight.restconf.nb.rfc8040.streams.WebSocketInitializer;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -51,13 +62,14 @@ public class CommunityRestConf extends AbstractLightyModule {
     private Server jettyServer;
     private LightyServerBuilder lightyServerBuilder;
     private SchemaContextHandler schemaCtxHandler;
+    private final ScheduledThreadPool scheduledThreadPool;
 
     public CommunityRestConf(final DOMDataBroker domDataBroker, final DOMRpcService domRpcService,
             final DOMActionService domActionService, final DOMNotificationService domNotificationService,
             final DOMMountPointService domMountPointService,
             final DOMSchemaService domSchemaService, final InetAddress inetAddress,
             final int httpPort, final String restconfServletContextPath,
-            final LightyServerBuilder serverBuilder) {
+            final LightyServerBuilder serverBuilder, final ScheduledThreadPool threadPool) {
         this.domDataBroker = domDataBroker;
         this.domRpcService = domRpcService;
         this.domActionService = domActionService;
@@ -68,16 +80,17 @@ public CommunityRestConf(final DOMDataBroker domDataBroker, final DOMRpcService
         this.httpPort = httpPort;
         this.inetAddress = inetAddress;
         this.restconfServletContextPath = restconfServletContextPath;
+        this.scheduledThreadPool = threadPool;
     }
 
     public CommunityRestConf(final DOMDataBroker domDataBroker,
             final DOMRpcService domRpcService, final DOMActionService domActionService,
             final DOMNotificationService domNotificationService, final DOMMountPointService domMountPointService,
             final DOMSchemaService domSchemaService, final InetAddress inetAddress, final int httpPort,
-            final String restconfServletContextPath) {
+            final String restconfServletContextPath, final ScheduledThreadPool threadPool) {
         this(domDataBroker, domRpcService, domActionService, domNotificationService,
                 domMountPointService, domSchemaService, inetAddress, httpPort,
-                restconfServletContextPath, null);
+                restconfServletContextPath, null, threadPool);
     }
 
     @Override
@@ -93,10 +106,15 @@ protected boolean initProcedure() {
         final RestconfApplication restconfApplication = new RestconfApplication(databindProvider,
                 this.domMountPointService, this.domDataBroker, this.domRpcService, this.domActionService,
                 this.domNotificationService, this.domSchemaService, streamsConfiguration);
+        final DataStreamApplication dataStreamApplication = new DataStreamApplication(databindProvider, this.domMountPointService,
+            new RestconfDataStreamServiceImpl(scheduledThreadPool, streamsConfiguration));
         final ServletContainer servletContainer8040 = new ServletContainer(ResourceConfig
                 .forApplication(restconfApplication));
         final ServletHolder jaxrs = new ServletHolder(servletContainer8040);
 
+        final ServletContainer dataStreamServletContainer = new ServletContainer(ResourceConfig.forApplication(dataStreamApplication));
+        final ServletHolder dataStreamHolder = new ServletHolder(dataStreamServletContainer);
+
         LOG.info("RestConf init complete, starting Jetty");
         LOG.info("http address:port {}:{}, url prefix: {}", this.inetAddress.toString(), this.httpPort,
                 this.restconfServletContextPath);
@@ -108,6 +126,19 @@ protected boolean initProcedure() {
                     new ServletContextHandler(contexts, this.restconfServletContextPath, true, false);
             mainHandler.addServlet(jaxrs, "/*");
 
+            final ServletContextHandler notifHandler =
+                    new ServletContextHandler(contexts, "/" + NOTIF, true, false);
+            notifHandler.addServlet(dataStreamHolder, "/*");
+
+            final ServletContextHandler wsHandler = new ServletContextHandler(contexts, "/" + DATA_SUBSCRIPTION, true, false);
+            final ServletContextHandler wsHandler1 = new ServletContextHandler(contexts, "/" + NOTIFICATION_STREAM, true, false);
+            final WebSocketInitializer webSocketInitializer = new WebSocketInitializer(scheduledThreadPool, streamsConfiguration);
+            ServletContext context = wsHandler1.getServletContext();
+            WebSocketServletFactory factory = new WebSocketServerFactory(context);
+            webSocketInitializer.configure(factory);
+            wsHandler.addServlet(new ServletHolder(webSocketInitializer), "/*");
+            wsHandler1.addServlet(new ServletHolder(webSocketInitializer), "/*");
+
             final ServletContextHandler rrdHandler =
                     new ServletContextHandler(contexts, "/.well-known", true, false);
             final RootFoundApplication rootDiscoveryApp = new RootFoundApplication(restconfServletContextPath);

lighty-modules/lighty-restconf-nb-community/src/main/java/io/lighty/modules/northbound/restconf/community/impl/CommunityRestConfBuilder.java

@@ -9,6 +9,7 @@
 
 import io.lighty.modules.northbound.restconf.community.impl.config.RestConfConfiguration;
 import io.lighty.server.LightyServerBuilder;
+import org.opendaylight.controller.config.threadpool.ScheduledThreadPool;
 
 /**
  * Builder for {@link CommunityRestConf}.
@@ -17,6 +18,7 @@ public final class CommunityRestConfBuilder {
 
     private RestConfConfiguration restconfConfiguration = null;
     private LightyServerBuilder lightyServerBuilder = null;
+    private ScheduledThreadPool threadPool = null;
 
 
     private CommunityRestConfBuilder(final RestConfConfiguration configuration) {
@@ -44,6 +46,19 @@ public CommunityRestConfBuilder withLightyServer(final LightyServerBuilder serve
         return this;
     }
 
+
+    /**
+     * Inject lighty server builder.
+     *
+     * @param pool input server builder.
+     * @return instance of {@link CommunityRestConfBuilder}.
+     */
+    public CommunityRestConfBuilder withScheduledThreadPool(final ScheduledThreadPool pool) {
+        this.threadPool = pool;
+        return this;
+    }
+
+
     /**
      * Build new {@link CommunityRestConf} instance from {@link CommunityRestConfBuilder}.
      * @return instance of CommunityRestConf.
@@ -55,6 +70,6 @@ public CommunityRestConf build() {
             this.restconfConfiguration.getDomMountPointService(),
             this.restconfConfiguration.getDomSchemaService(),
             this.restconfConfiguration.getInetAddress(), this.restconfConfiguration.getHttpPort(),
-            this.restconfConfiguration.getRestconfServletContextPath(), this.lightyServerBuilder);
+            this.restconfConfiguration.getRestconfServletContextPath(), this.lightyServerBuilder, this.threadPool);
     }
 }