Merge "Implement framework to process REST notifications"

7 files changed, 380 insertions, 0 deletions

diff --git a/fabric-discovery-plugin/provider/src/main/java/org/onap/ccsdk/sli/plugins/fabricdiscovery/ClientMessageCallback.java b/fabric-discovery-plugin/provider/src/main/java/org/onap/ccsdk/sli/plugins/fabricdiscovery/ClientMessageCallback.java
new file mode 100644
index 000000000..0a8ba9e75
--- /dev/null
+++ b/fabric-discovery-plugin/provider/src/main/java/org/onap/ccsdk/sli/plugins/fabricdiscovery/ClientMessageCallback.java
@@ -0,0 +1,20 @@
+package org.onap.ccsdk.sli.plugins.fabricdiscovery;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
+
+/**
+ * Created by arun on 9/9/17.
+ */
+public class ClientMessageCallback implements IClientMessageCallback {
+  private static final Logger LOG = LoggerFactory.getLogger(ClientMessageCallback.class);
+
+  @Override
+  public void onMessageReceived(final Object message) {
+    if (message instanceof TextWebSocketFrame) {
+      LOG.info("received message {}" + ((TextWebSocketFrame) message).text());
+    }
+  }
+
+}
diff --git a/fabric-discovery-plugin/provider/src/main/java/org/onap/ccsdk/sli/plugins/fabricdiscovery/FabricDiscoveryPlugin.java b/fabric-discovery-plugin/provider/src/main/java/org/onap/ccsdk/sli/plugins/fabricdiscovery/FabricDiscoveryPlugin.java
new file mode 100644
index 000000000..7a0c68f41
--- /dev/null
+++ b/fabric-discovery-plugin/provider/src/main/java/org/onap/ccsdk/sli/plugins/fabricdiscovery/FabricDiscoveryPlugin.java
@@ -0,0 +1,129 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * openECOMP : SDN-C
+ * ================================================================================
+ * Copyright (C) 2017 AT&T Intellectual Property. All rights
+ *             reserved.
+ * ================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * 
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.ccsdk.sli.plugins.fabricdiscovery;
+
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import org.onap.ccsdk.sli.core.sli.SvcLogicContext;
+import org.onap.ccsdk.sli.core.sli.SvcLogicException;
+import org.onap.ccsdk.sli.core.sli.SvcLogicJavaPlugin;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+public class FabricDiscoveryPlugin implements SvcLogicJavaPlugin, IFabricDiscoveryService {
+
+    private ExecutorService service;
+    private Map<String, WebSocketClient> streamMap;
+    private static final Logger LOG = LoggerFactory.getLogger(FabricDiscoveryPlugin.class);
+
+    public FabricDiscoveryPlugin() {
+        service = Executors.newFixedThreadPool(10);
+        streamMap = new ConcurrentHashMap<String, WebSocketClient> ();
+    }
+
+    @Override
+    public void processDcNotificationStream (Map<String, String> paramMap, SvcLogicContext ctx) throws SvcLogicException {
+        boolean enable = Boolean.parseBoolean(parseParam(paramMap, "enable", true, null));
+        String stream = parseParam(paramMap, "stream", true, null);
+
+        LOG.info("{} monitoring notification stream: {}", (enable) ? "START" : "STOP", stream);
+        try {
+            service.execute(new Runnable () {
+                public void run () {
+                    try {
+                        URI uri = new URI(stream);
+                        if (enable) {
+                            if (streamMap.get(stream) != null) {
+                                LOG.info("Notification Stream: {} is already being monitoried", stream);
+                                return;
+                            }
+                            IClientMessageCallback messageCallback = new ClientMessageCallback();
+                            WebSocketClient wcClient = new WebSocketClient(uri, messageCallback);
+                            streamMap.put(stream, wcClient);
+                            wcClient.initialize();
+                            try {
+                                wcClient.connect();
+                            } catch (InterruptedException e) {
+                                LOG.info("Web Socket Client throws Exception: ", e.getMessage());
+                            }
+                        } else {
+                            WebSocketClient wc = streamMap.get(stream);
+                            if (wc != null) {
+                                try {
+                                    wc.close("Closing");
+                                } catch (InterruptedException e) {
+                                    LOG.info("Web Socket Client throws Exception: ", e.getMessage());
+                                }
+                            }
+                        }
+                    } catch (URISyntaxException e) {
+                        LOG.info("Exception converting stream to URI with: ", e.getMessage());
+                    }
+                }
+            });
+        } catch (Exception e) {
+            LOG.info("Web Socket client connection throws an exception: ", e.getMessage());
+        }
+    }
+
+    private String parseParam(Map<String, String> paramMap, String name, boolean required, String def)
+        throws SvcLogicException {
+        String s = paramMap.get(name);
+
+        if (s == null || s.trim().length() == 0) {
+            if (!required)
+                return def;
+            throw new SvcLogicException("Parameter " + name + " is required in PropertiesNode");
+        }
+
+        s = s.trim();
+        String value = "";
+        int i = 0;
+        int i1 = s.indexOf('%');
+        while (i1 >= 0) {
+            int i2 = s.indexOf('%', i1 + 1);
+            if (i2 < 0)
+                throw new SvcLogicException("Cannot parse parameter " + name + ": " + s + ": no matching %");
+
+            String varName = s.substring(i1 + 1, i2);
+            String varValue = System.getenv(varName);
+            if (varValue == null)
+                varValue = "";
+
+            value = (new StringBuilder()).append(value)
+                    .append(s.substring(i, i1))
+                    .append(varValue).toString();
+            i = i2 + 1;
+            i1 = s.indexOf('%', i);
+        }
+        value = (new StringBuilder()).append(value)
+                     .append(s.substring(i)).toString();
+
+        LOG.info("Parameter {}: {}", name, value);
+        return value;
+    }
+}
diff --git a/fabric-discovery-plugin/provider/src/main/java/org/onap/ccsdk/sli/plugins/fabricdiscovery/IClientMessageCallback.java b/fabric-discovery-plugin/provider/src/main/java/org/onap/ccsdk/sli/plugins/fabricdiscovery/IClientMessageCallback.java
new file mode 100644
index 000000000..e555d3382
--- /dev/null
+++ b/fabric-discovery-plugin/provider/src/main/java/org/onap/ccsdk/sli/plugins/fabricdiscovery/IClientMessageCallback.java
@@ -0,0 +1,16 @@
+/*
+ * Copyright (c) 2014, 2015 Cisco Systems, Inc. and others.  All rights reserved.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License v1.0 which accompanies this distribution,
+ * and is available at http://www.eclipse.org/legal/epl-v10.html
+ */
+package org.onap.ccsdk.sli.plugins.fabricdiscovery;
+
+/**
+ * Created by mbobak on 1/22/14.
+ */
+public interface IClientMessageCallback {
+
+    void onMessageReceived(Object message);
+}
diff --git a/fabric-discovery-plugin/provider/src/main/java/org/onap/ccsdk/sli/plugins/fabricdiscovery/IFabricDiscoveryService.java b/fabric-discovery-plugin/provider/src/main/java/org/onap/ccsdk/sli/plugins/fabricdiscovery/IFabricDiscoveryService.java
new file mode 100644
index 000000000..857d75a99
--- /dev/null
+++ b/fabric-discovery-plugin/provider/src/main/java/org/onap/ccsdk/sli/plugins/fabricdiscovery/IFabricDiscoveryService.java
@@ -0,0 +1,17 @@
+package org.onap.ccsdk.sli.plugins.fabricdiscovery;
+
+import java.util.Map;
+import org.onap.ccsdk.sli.core.sli.SvcLogicContext;
+import org.onap.ccsdk.sli.core.sli.SvcLogicException;
+
+/**
+ * Created by arun on 9/10/17.
+ */
+public interface IFabricDiscoveryService {
+
+  /* (non-Javadoc)
+   * @see void#processDcNotificationStream(java.lang.String, java.lang.Integer, java.lang.String, java.lang.Boolean)
+   */
+  void processDcNotificationStream (Map<String, String> paramMap, SvcLogicContext ctx) throws SvcLogicException;
+
+}
diff --git a/fabric-discovery-plugin/provider/src/main/java/org/onap/ccsdk/sli/plugins/fabricdiscovery/WebSocketClient.java b/fabric-discovery-plugin/provider/src/main/java/org/onap/ccsdk/sli/plugins/fabricdiscovery/WebSocketClient.java
new file mode 100644
index 000000000..bdecdc742
--- /dev/null
+++ b/fabric-discovery-plugin/provider/src/main/java/org/onap/ccsdk/sli/plugins/fabricdiscovery/WebSocketClient.java
@@ -0,0 +1,96 @@
+/*
+ * Copyright (c) 2013 Cisco Systems, Inc. and others.  All rights reserved.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License v1.0 which accompanies this distribution,
+ * and is available at http://www.eclipse.org/legal/epl-v10.html
+ */
+package org.onap.ccsdk.sli.plugins.fabricdiscovery;
+
+import io.netty.bootstrap.Bootstrap;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.ChannelPipeline;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.socket.SocketChannel;
+import io.netty.channel.socket.nio.NioSocketChannel;
+import io.netty.handler.codec.http.HttpClientCodec;
+import io.netty.handler.codec.http.HttpObjectAggregator;
+import io.netty.handler.codec.http.websocketx.CloseWebSocketFrame;
+import io.netty.handler.codec.http.websocketx.PingWebSocketFrame;
+import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
+import io.netty.handler.codec.http.websocketx.WebSocketClientHandshakerFactory;
+import io.netty.handler.codec.http.websocketx.WebSocketVersion;
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.net.URI;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class WebSocketClient {
+
+    private static final Logger LOG = LoggerFactory.getLogger(WebSocketClient.class);
+
+    private final URI uri;
+    private final Bootstrap bootstrap = new Bootstrap();
+    private final WebSocketClientHandler clientHandler;
+    private Channel clientChannel;
+    private final EventLoopGroup group = new NioEventLoopGroup();
+
+    public WebSocketClient(final URI uri, final IClientMessageCallback clientMessageCallback) {
+        this.uri = uri;
+        clientHandler = new WebSocketClientHandler(WebSocketClientHandshakerFactory.newHandshaker(uri,
+                WebSocketVersion.V13, null, false, null), clientMessageCallback);
+        // last null could be replaced with DefaultHttpHeaders
+        initialize();
+    }
+
+    public void initialize() {
+
+        String protocol = uri.getScheme();
+        if (!"ws".equals(protocol)) {
+            throw new IllegalArgumentException("Unsupported protocol: " + protocol);
+        }
+
+        bootstrap.group(group).channel(NioSocketChannel.class).handler(new ChannelInitializer<SocketChannel>() {
+            @Override
+            public void initChannel(final SocketChannel ch) throws Exception {
+                ChannelPipeline pipeline = ch.pipeline();
+                pipeline.addLast("http-codec", new HttpClientCodec());
+                pipeline.addLast("aggregator", new HttpObjectAggregator(8192));
+                pipeline.addLast("ws-handler", clientHandler);
+            }
+        });
+    }
+
+    public void connect() throws InterruptedException {
+        LOG.info("WebSocket Client connecting");
+        clientChannel = bootstrap.connect(uri.getHost(), uri.getPort()).sync().channel();
+        clientHandler.handshakeFuture().sync();
+    }
+
+    public void writeAndFlush(final String message) {
+        clientChannel.writeAndFlush(new TextWebSocketFrame(message));
+    }
+
+    public void writeAndFlush(final Object message) {
+        clientChannel.writeAndFlush(message);
+    }
+
+    public void ping() {
+        clientChannel.writeAndFlush(new PingWebSocketFrame(Unpooled.copiedBuffer(new byte[] { 1, 2, 3, 4, 5, 6 })));
+    }
+
+    public void close(final String reasonText) throws InterruptedException {
+        CloseWebSocketFrame closeWebSocketFrame = new CloseWebSocketFrame(1000, reasonText);
+        clientChannel.writeAndFlush(closeWebSocketFrame);
+
+        // WebSocketClientHandler will close the connection when the server
+        // responds to the CloseWebSocketFrame.
+        clientChannel.closeFuture().sync();
+        group.shutdownGracefully();
+    }
+}
diff --git a/fabric-discovery-plugin/provider/src/main/java/org/onap/ccsdk/sli/plugins/fabricdiscovery/WebSocketClientHandler.java b/fabric-discovery-plugin/provider/src/main/java/org/onap/ccsdk/sli/plugins/fabricdiscovery/WebSocketClientHandler.java
new file mode 100644
index 000000000..c1c5dc30f
--- /dev/null
+++ b/fabric-discovery-plugin/provider/src/main/java/org/onap/ccsdk/sli/plugins/fabricdiscovery/WebSocketClientHandler.java
@@ -0,0 +1,92 @@
+/*
+ * Copyright (c) 2013 Cisco Systems, Inc. and others.  All rights reserved.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License v1.0 which accompanies this distribution,
+ * and is available at http://www.eclipse.org/legal/epl-v10.html
+ */
+package org.onap.ccsdk.sli.plugins.fabricdiscovery;
+
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelPromise;
+import io.netty.channel.SimpleChannelInboundHandler;
+import io.netty.handler.codec.http.FullHttpResponse;
+import io.netty.handler.codec.http.websocketx.CloseWebSocketFrame;
+import io.netty.handler.codec.http.websocketx.PongWebSocketFrame;
+import io.netty.handler.codec.http.websocketx.WebSocketClientHandshaker;
+import io.netty.handler.codec.http.websocketx.WebSocketFrame;
+import io.netty.util.CharsetUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class WebSocketClientHandler extends SimpleChannelInboundHandler<Object> {
+
+    private static final Logger LOG = LoggerFactory.getLogger(WebSocketClientHandler.class.toString());
+    private final WebSocketClientHandshaker handshaker;
+    private ChannelPromise handshakeFuture;
+    private final IClientMessageCallback messageListener;
+
+    public WebSocketClientHandler(WebSocketClientHandshaker handshaker, IClientMessageCallback listener) {
+        this.handshaker = handshaker;
+        this.messageListener = listener;
+    }
+
+    public ChannelFuture handshakeFuture() {
+        return handshakeFuture;
+    }
+
+    @Override
+    public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
+        handshakeFuture = ctx.newPromise();
+    }
+
+    @Override
+    public void channelActive(ChannelHandlerContext ctx) throws Exception {
+        handshaker.handshake(ctx.channel());
+    }
+
+    @Override
+    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
+        LOG.info("WebSocket Client disconnected!");
+    }
+
+    @Override
+    public void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {
+        Channel ch = ctx.channel();
+        if (!handshaker.isHandshakeComplete()) {
+            handshaker.finishHandshake(ch, (FullHttpResponse) msg);
+            LOG.info("WebSocket Client connected!");
+            handshakeFuture.setSuccess();
+            return;
+        }
+
+        if (msg instanceof FullHttpResponse) {
+            FullHttpResponse response = (FullHttpResponse) msg;
+            throw new Exception("Unexpected FullHttpResponse (getStatus=" + response.getStatus() + ", content="
+                    + response.content().toString(CharsetUtil.UTF_8) + ')');
+        }
+
+        messageListener.onMessageReceived(msg);
+        WebSocketFrame frame = (WebSocketFrame) msg;
+
+        if (frame instanceof PongWebSocketFrame) {
+            LOG.info("WebSocket Client received pong");
+        } else if (frame instanceof CloseWebSocketFrame) {
+            LOG.info("WebSocket Client received closing");
+            ch.close();
+        }
+    }
+
+    @Override
+    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
+        LOG.info("Cause: {} .", cause.toString());
+
+        if (!handshakeFuture.isDone()) {
+            handshakeFuture.setFailure(cause);
+        }
+
+        ctx.close();
+    }
+}
diff --git a/fabric-discovery-plugin/provider/src/main/resources/org/opendaylight/blueprint/fabricdiscoveryplugin-blueprint.xml b/fabric-discovery-plugin/provider/src/main/resources/org/opendaylight/blueprint/fabricdiscoveryplugin-blueprint.xml
new file mode 100644
index 000000000..43137bb89
--- /dev/null
+++ b/fabric-discovery-plugin/provider/src/main/resources/org/opendaylight/blueprint/fabricdiscoveryplugin-blueprint.xml
@@ -0,0 +1,10 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<blueprint xmlns="http://www.osgi.org/xmlns/blueprint/v1.0.0"
+           xmlns:odl="http://opendaylight.org/xmlns/blueprint/v1.0.0"
+           odl:use-default-for-reference-types="true">
+
+
+    <bean id="fabricDiscoveryPlugin" class="org.onap.ccsdk.sli.plugins.fabricdiscovery.FabricDiscoveryPlugin" />
+    <service ref="fabricDiscoveryPlugin" interface="org.onap.ccsdk.sli.plugins.fabricdiscovery.FabricDiscoveryPlugin" />
+
+</blueprint>