diff options
Diffstat (limited to 'fabric-discovery-plugin/provider/src/main')
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> |