aboutsummaryrefslogtreecommitdiffstats
path: root/fabric-discovery-plugin/provider/src
diff options
context:
space:
mode:
Diffstat (limited to 'fabric-discovery-plugin/provider/src')
-rw-r--r--fabric-discovery-plugin/provider/src/main/java/org/onap/ccsdk/sli/plugins/fabricdiscovery/ClientMessageCallback.java20
-rw-r--r--fabric-discovery-plugin/provider/src/main/java/org/onap/ccsdk/sli/plugins/fabricdiscovery/FabricDiscoveryPlugin.java129
-rw-r--r--fabric-discovery-plugin/provider/src/main/java/org/onap/ccsdk/sli/plugins/fabricdiscovery/IClientMessageCallback.java16
-rw-r--r--fabric-discovery-plugin/provider/src/main/java/org/onap/ccsdk/sli/plugins/fabricdiscovery/IFabricDiscoveryService.java17
-rw-r--r--fabric-discovery-plugin/provider/src/main/java/org/onap/ccsdk/sli/plugins/fabricdiscovery/WebSocketClient.java96
-rw-r--r--fabric-discovery-plugin/provider/src/main/java/org/onap/ccsdk/sli/plugins/fabricdiscovery/WebSocketClientHandler.java92
-rw-r--r--fabric-discovery-plugin/provider/src/main/resources/org/opendaylight/blueprint/fabricdiscoveryplugin-blueprint.xml10
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 00000000..0a8ba9e7
--- /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 00000000..7a0c68f4
--- /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 00000000..e555d338
--- /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 00000000..857d75a9
--- /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 00000000..bdecdc74
--- /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 00000000..c1c5dc30
--- /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 00000000..43137bb8
--- /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>