fabric-discovery-plugin/provider/src/main/java/org/onap/ccsdk/sli/plugins/fabricdiscovery/FabricDiscoveryPlugin.java

   1 /*-
   2  * ============LICENSE_START=======================================================
   3  * openECOMP : SDN-C
   4  * ================================================================================
   5  * Copyright (C) 2017 AT&T Intellectual Property. All rights
   6  *             reserved.
   7  * ================================================================================
   8  * Licensed under the Apache License, Version 2.0 (the "License");
   9  * you may not use this file except in compliance with the License.
  10  * You may obtain a copy of the License at
  11  *
  12  *      http://www.apache.org/licenses/LICENSE-2.0
  13  *
  14  * Unless required by applicable law or agreed to in writing, software
  15  * distributed under the License is distributed on an "AS IS" BASIS,
  16  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  17  * See the License for the specific language governing permissions and
  18  * limitations under the License.
  19  * ============LICENSE_END=========================================================
  20  */
  21
  22 package org.onap.ccsdk.sli.plugins.fabricdiscovery;
  23
  24 import java.net.URI;
  25 import java.net.URISyntaxException;
  26 import java.util.Map;
  27 import java.util.concurrent.ConcurrentHashMap;
  28 import java.util.concurrent.ExecutorService;
  29 import java.util.concurrent.Executors;
  30 import org.onap.ccsdk.sli.core.sli.SvcLogicContext;
  31 import org.onap.ccsdk.sli.core.sli.SvcLogicException;
  32 import org.onap.ccsdk.sli.core.sli.SvcLogicJavaPlugin;
  33 import org.slf4j.Logger;
  34 import org.slf4j.LoggerFactory;
  35
  36
  37 public class FabricDiscoveryPlugin implements SvcLogicJavaPlugin, IFabricDiscoveryService {
  38
  39     private ExecutorService service;
  40     private Map<String, WebSocketClient> streamMap;
  41     private static final Logger LOG = LoggerFactory.getLogger(FabricDiscoveryPlugin.class);
  42     private static final String STREAM_PREFIX = "ws://";
  43     private static final String FB_DISCOVERY_STATUS = "fb-response";
  44
  45     public FabricDiscoveryPlugin() {
  46         service = Executors.newFixedThreadPool(10);
  47         streamMap = new ConcurrentHashMap<String, WebSocketClient> ();
  48     }
  49
  50     @Override
  51     public void processDcNotificationStream (Map<String, String> paramMap, SvcLogicContext ctx) throws SvcLogicException {
  52         boolean enable;
  53         String stream = parseParam(paramMap, "stream", true, null);
  54         String prefix = parseParam(paramMap, "contextPrefix", false, null);
  55         String enableStr = parseParam(paramMap, "enable", true, null);
  56
  57         // Validate the input parameters
  58         String pfx = (prefix != null) ? prefix + '.' : "";
  59         if ("true".equalsIgnoreCase(enableStr)) {
  60             enable = true;
  61         } else if ("false".equalsIgnoreCase(enableStr)) {
  62             enable = false;
  63         } else {
  64             ctx.setAttribute(pfx + FB_DISCOVERY_STATUS, "Failure");
  65             throw new SvcLogicException("Incorrect parameter: enable. Valid values are ['true', 'false']");
  66         }
  67         if (!STREAM_PREFIX.equalsIgnoreCase(stream.substring(0, 5))) {
  68             ctx.setAttribute(pfx + FB_DISCOVERY_STATUS, "Failure");
  69             throw new SvcLogicException("Incorrect parameter: stream, Input is not a web socket address");
  70         }
  71
  72         ctx.setAttribute(pfx + FB_DISCOVERY_STATUS, "Success");
  73         LOG.info("{} monitoring notification stream: {}", (enable) ? "START" : "STOP", stream);
  74
  75         try {
  76             service.execute(new Runnable () {
  77                 public void run () {
  78                     try {
  79                         URI uri = new URI(stream);
  80                         if (enable) {
  81                             if (streamMap.get(stream) != null) {
  82                                 LOG.info("Notification Stream: {} is already being monitoried", stream);
  83                                 return;
  84                             }
  85                             IClientMessageCallback messageCallback = new ClientMessageCallback();
  86                             WebSocketClient wcClient = new WebSocketClient(uri, messageCallback);
  87                             streamMap.put(stream, wcClient);
  88                             wcClient.initialize();
  89                             try {
  90                                 wcClient.connect();
  91                             } catch (InterruptedException e) {
  92                                 LOG.info("Web Socket Client throws Exception: ", e.getMessage());
  93                             }
  94                         } else {
  95                             WebSocketClient wc = streamMap.get(stream);
  96                             if (wc != null) {
  97                                 try {
  98                                     wc.close("Closing");
  99                                 } catch (InterruptedException e) {
 100                                     LOG.info("Web Socket Client throws Exception: ", e.getMessage());
 101                                 }
 102                             }
 103                         }
 104                     } catch (URISyntaxException e) {
 105                         LOG.info("Exception converting stream to URI with: ", e.getMessage());
 106                     }
 107                 }
 108             });
 109         } catch (Exception e) {
 110             LOG.info("Web Socket client connection throws an exception: ", e.getMessage());
 111         }
 112     }
 113
 114     private String parseParam(Map<String, String> paramMap, String name, boolean required, String def)
 115         throws SvcLogicException {
 116         String s = paramMap.get(name);
 117
 118         if (s == null || s.trim().length() == 0) {
 119             if (!required)
 120                 return def;
 121             throw new SvcLogicException("Parameter " + name + " is required in PropertiesNode");
 122         }
 123
 124         s = s.trim();
 125         String value = "";
 126         int i = 0;
 127         int i1 = s.indexOf('%');
 128         while (i1 >= 0) {
 129             int i2 = s.indexOf('%', i1 + 1);
 130             if (i2 < 0)
 131                 throw new SvcLogicException("Cannot parse parameter " + name + ": " + s + ": no matching %");
 132
 133             String varName = s.substring(i1 + 1, i2);
 134             String varValue = System.getenv(varName);
 135             if (varValue == null)
 136                 varValue = "";
 137
 138             value = (new StringBuilder()).append(value)
 139                     .append(s.substring(i, i1))
 140                     .append(varValue).toString();
 141             i = i2 + 1;
 142             i1 = s.indexOf('%', i);
 143         }
 144         value = (new StringBuilder()).append(value)
 145                      .append(s.substring(i)).toString();
 146
 147         LOG.info("Parameter {}: {}", name, value);
 148         return value;
 149     }
 150 }