diff options
4 files changed, 726 insertions, 0 deletions
diff --git a/plugins/restconf-client/provider/src/main/java/org/onap/ccsdk/sli/plugins/restconfdiscovery/PeriodicDiscoveryNode.java b/plugins/restconf-client/provider/src/main/java/org/onap/ccsdk/sli/plugins/restconfdiscovery/PeriodicDiscoveryNode.java new file mode 100644 index 000000000..220de8790 --- /dev/null +++ b/plugins/restconf-client/provider/src/main/java/org/onap/ccsdk/sli/plugins/restconfdiscovery/PeriodicDiscoveryNode.java @@ -0,0 +1,681 @@ +/*- + * ============LICENSE_START======================================================= + * ONAP - CCSDK + * ================================================================================ + * Copyright (C) 2018 Huawei Technologies Co., Ltd. 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.restconfdiscovery; + +import com.google.common.collect.ImmutableMap; +import org.glassfish.jersey.media.sse.EventInput; +import org.glassfish.jersey.media.sse.EventSource; +import org.glassfish.jersey.media.sse.InboundEvent; +import org.glassfish.jersey.media.sse.SseFeature; +import org.onap.ccsdk.sli.core.sli.SvcLogicContext; +import org.onap.ccsdk.sli.core.sli.SvcLogicException; +import org.onap.ccsdk.sli.core.utils.common.AcceptIpAddressHostNameVerifier; +import org.onap.ccsdk.sli.plugins.restapicall.Parameters; +import org.onap.ccsdk.sli.plugins.restapicall.RestapiCallNode; +import org.onap.ccsdk.sli.plugins.restconfapicall.RestconfApiCallNode; +import org.slf4j.Logger; + +import javax.net.ssl.SSLContext; +import javax.net.ssl.TrustManager; +import javax.net.ssl.X509TrustManager; +import javax.ws.rs.client.Client; +import javax.ws.rs.client.ClientBuilder; +import javax.ws.rs.client.WebTarget; +import javax.ws.rs.core.Response; +import java.net.MalformedURLException; +import java.net.URL; +import java.security.KeyManagementException; +import java.security.NoSuchAlgorithmException; +import java.security.cert.CertificateException; +import java.security.cert.X509Certificate; +import java.util.Map; +import java.util.NoSuchElementException; +import java.util.concurrent.*; + +import static org.onap.ccsdk.sli.plugins.restapicall.JsonParser.convertToProperties; +import static org.slf4j.LoggerFactory.getLogger; + +/** + * Representation of a plugin to subscribe for notification and then + * to handle the received notifications. + */ +public class PeriodicDiscoveryNode implements RestConfSBController, SvcLogicDiscoveryPlugin { + + private static final Logger log = getLogger(PeriodicDiscoveryNode.class); + + private static final String ROOT_RESOURCE = "/restconf"; + private static final String SUBSCRIBER_ID = "subscriberId"; + private static final String RESPONSE_CODE = "response-code"; + private static final String RESPONSE_PREFIX = "responsePrefix"; + private static final String OUTPUT_IDENTIFIER = "ietf-subscribed-notif" + + "ications:establish-subscription.output.identifier"; + private static final String OUTPUT_IDENTIFIER_NO_PREFIX = "output.identifier"; + private static final String RESPONSE_CODE_200 = "200"; + private static final String SSE_URL = "sseConnectURL"; + private static final String PERIODIC_PUL_URL = "periodicPullURL"; + private static final String REST_API_URL = "restapiUrl"; + private static final String RESOURCE_PATH_PREFIX = "/data/"; + private static final String NOTIFICATION_PATH_PREFIX = "/streams/"; + private static final String DEVICE_IP = "deviceIp"; + private static final String DEVICE_PORT = "devicePort"; + private static final String DOUBLESLASH = "//"; + private static final String COLON = ":"; + + private RestconfApiCallNode restconfApiCallNode; + private RestapiCallNode restapiCallNode = new RestapiCallNode(); + private volatile Map<String, SubscriptionInfo> subscriptionInfoMap = new ConcurrentHashMap<>(); + private volatile LinkedBlockingQueue<String> eventQueue = new LinkedBlockingQueue<>(); + private Map<DeviceId, PeriodicPullRunnable> periodicRunnableTable = new ConcurrentHashMap<>(); + private Map<DeviceId, String> subscribedDevicesTable = new ConcurrentHashMap<>(); + private Map<DeviceId, BlockingQueue<String>> eventQMap = new ConcurrentHashMap<>(); + private Map<DeviceId, InternalPeriodicPullingProcessorRunnable> + processorRunnableTable = new ConcurrentHashMap<>(); + private final Map<DeviceId, RestSBDevice> deviceMap = new ConcurrentHashMap<>(); + private final Map<DeviceId, Client> clientMap = new ConcurrentHashMap<>(); + private ExecutorService executor = Executors.newCachedThreadPool(); + private ScheduledExecutorService scheduledExecutor = Executors.newScheduledThreadPool(2); + + /** + * Creates an instance of RestconfDiscoveryNode and starts processing of + * event. + * + * @param r restconf api call node + */ + public PeriodicDiscoveryNode(RestconfApiCallNode r) { + log.info("inside RestconfDiscoveryNode Constructor"); + this.restconfApiCallNode = r; + this.activate(); +// ExecutorService e = Executors.newFixedThreadPool(20); +// EventProcessor p = new EventProcessor(this); +// for (int i = 0; i < 20; ++i) { +// e.execute(p); +// } + } + + public void activate() { + log.info("RESTCONF SBI Started"); + } + + public void deactivate() { + log.info("RESTCONF SBI Stopped"); + executor.shutdown(); + this.getClientMap().clear(); + this.getDeviceMap().clear(); + } + + public Map<DeviceId, RestSBDevice> getDeviceMap() { + return deviceMap; + } + + public Map<DeviceId, Client> getClientMap() { + return clientMap; + } + + @Override + public Map<DeviceId, RestSBDevice> getDevices() { + log.trace("RESTCONF SBI::getDevices"); + return ImmutableMap.copyOf(deviceMap); + } + + @Override + public RestSBDevice getDevice(DeviceId deviceInfo) { + log.trace("RESTCONF SBI::getDevice with deviceId"); + return deviceMap.get(deviceInfo); + } + + @Override + public RestSBDevice getDevice(String ip, int port) { + log.trace("RESTCONF SBI::getDevice with ip and port"); + try { + if (!deviceMap.isEmpty()) { + return deviceMap.values().stream().filter(v -> v.ip().equals(ip) && v.port() == port).findFirst().get(); + } + } catch (NoSuchElementException noSuchElementException) { + log.error("getDevice::device {}:{} does not exist in deviceMap", ip, port); + } + return null; + } + + @Override + public void addDevice(RestSBDevice device) { + log.trace("RESTCONF SBI::addDevice"); + if (!deviceMap.containsKey(device.deviceId())) { + if (device.username() != null) { + String username = device.username(); + String password = device.password() == null ? "" : device.password(); + // authenticate(client, username, password); + } + BlockingQueue<String> newBlockingQueue = new LinkedBlockingQueue<>(); + eventQMap.put(device.deviceId(), newBlockingQueue); + InternalPeriodicPullingProcessorRunnable eventProcessorRunnable = + new InternalPeriodicPullingProcessorRunnable(device.deviceId()); + processorRunnableTable.put(device.deviceId(), eventProcessorRunnable); + log.trace("addDevice::restconf event processor runnable is created and is going for execute"); + executor.execute(eventProcessorRunnable); + log.trace("addDevice::restconf event processor runnable was sent for execute"); + deviceMap.put(device.deviceId(), device); + } else { + log.warn("addDevice::Trying to add a device which already exists {}", device.deviceId()); + } + } + + @Override + public void removeDevice(DeviceId deviceId) { + log.trace("RESTCONF SBI::removeDevice"); + eventQMap.remove(deviceId); + clientMap.remove(deviceId); + deviceMap.remove(deviceId); + } + + @Override + public void establishSubscription(Map<String, String> paramMap, + SvcLogicContext ctx) throws SvcLogicException { + String subscriberId = paramMap.get(SUBSCRIBER_ID); + if (subscriberId == null) { + throw new SvcLogicException("Subscriber Id is null"); + } + + restconfApiCallNode.sendRequest(paramMap, ctx); + + if (getResponseCode(paramMap.get(RESPONSE_PREFIX), ctx).equals(RESPONSE_CODE_200)) { + // TODO: save subscription id and subscriber in MYSQL + + establishPersistentConnection(paramMap, ctx, subscriberId); + } else { + log.info("Failed to subscribe {}", subscriberId); + throw new SvcLogicException(ctx.getAttribute(RESPONSE_CODE)); + } + } + + @Override + public void establishSubscriptionOnly(Map<String, String> paramMap, SvcLogicContext ctx) + throws SvcLogicException { + String subscriberId = paramMap.get(SUBSCRIBER_ID); + if (subscriberId == null) { + throw new SvcLogicException("Subscriber Id is null"); + } + + String subscribeUrlString = paramMap.get(REST_API_URL); + URL subscribeUrl = null; + RestSBDevice dev = null; + try { + subscribeUrl = new URL(subscribeUrlString); + dev = getDevice(subscribeUrl.getHost(), subscribeUrl.getPort()); + } catch (MalformedURLException e) { + log.error("establishSubscriptionOnly::MalformedURLException happened. e: {}", e); + return; + } + + if (dev == null) { + log.warn("establishSubscriptionOnly::device does not exist in the map. Trying to create one now."); + //FIXME: TODO: create a new RestSBDevice and add it to the map, as well as a client and clientMap + dev = new DefaultRestSBDevice(subscribeUrl.getHost(), + subscribeUrl.getPort(), "onos", "rocks", "http", + subscribeUrl.getHost() + ":" + subscribeUrl.getPort(), true); + this.addDevice(dev); + } + + if (!subscribedDevicesTable.containsKey(dev.deviceId())) { + log.info("establishSubscriptionOnly::The device {} has not been subscribed yet. " + + "Trying to subscribe it now..."); + restapiCallNode.sendRequest(paramMap, ctx); + if (getResponseCode(paramMap.get(RESPONSE_PREFIX), ctx).equals(RESPONSE_CODE_200)) { + // TODO: save subscription id and subscriber in MYSQL + String id = getOutputIdentifierNoPrefix(paramMap.get(RESPONSE_PREFIX), ctx); + log.info("establishSubscriptionOnly::Subscription is done successfully and " + + "the output.identifier is: {}", id); + log.info("establishSubscriptionOnly::The subscriptionID returned by the server " + + "does not exist in the map. Adding it now..."); + subscribedDevicesTable.put(dev.deviceId(), id); + + SvcLogicGraphInfo callbackDG = new SvcLogicGraphInfo(paramMap.get("module"), + paramMap.get("rpc"), + paramMap.get("version"), + paramMap.get("mode")); + SubscriptionInfo info = new SubscriptionInfo(); + info.callBackDG(callbackDG); + info.subscriptionId(id); + info.subscriberId(subscriberId); + subscriptionInfoMap.put(id, info); + + } + } + + } + + @Override + public void modifySubscription(Map<String, String> paramMap, SvcLogicContext ctx) { + // TODO: to be implemented + } + + @Override + public void deleteSubscription(Map<String, String> paramMap, SvcLogicContext ctx) { + String id = getSubscriptionId(paramMap.get(SUBSCRIBER_ID)); + if (id != null) { + subscriptionInfoMap.remove(id); + } + } + + protected String getTokenId(String customHttpHeaders) { + if (customHttpHeaders.contains("=")) { + String[] s = customHttpHeaders.split("="); + return s[1]; + } + return customHttpHeaders; + } + + protected WebTarget addToken(WebTarget target, String customHttpHeaders) { + if (customHttpHeaders == null) { + return target; + } + + return new AdditionalHeaderWebTarget( + target, getTokenId(customHttpHeaders)); + } + + /** + * Establishes a persistent between the client and server. + * + * @param paramMap input paramter map + * @param ctx service logic context + * @param subscriberId subscriber identifier + */ + void establishPersistentConnection(Map<String, String> paramMap, SvcLogicContext ctx, + String subscriberId) { + } + + /** + * Returns response code. + * + * @param prefix prefix given in input parameter + * @param ctx service logic context + * @return response code + */ + String getResponseCode(String prefix, SvcLogicContext ctx) { + return ctx.getAttribute(getPrefix(prefix) + RESPONSE_CODE); + } + + String getOutputIdentifierNoPrefix(String prefix, SvcLogicContext ctx) { + return ctx.getAttribute(getPrefix(prefix) + OUTPUT_IDENTIFIER_NO_PREFIX); + } + + /** + * Returns subscription id from event. + * + * @param prefix prefix given in input parameter + * @param ctx service logic context + * @return subscription id from event + */ + String getOutputIdentifier(String prefix, SvcLogicContext ctx) { + return ctx.getAttribute(getPrefix(prefix) + OUTPUT_IDENTIFIER); + } + + private String getPrefix(String prefix) { + return prefix != null ? prefix + "." : ""; + } + + private String getSubscriptionId(String subscriberId) { + for (Map.Entry<String,SubscriptionInfo> entry + : subscriptionInfoMap.entrySet()) { + if (entry.getValue().subscriberId() + .equals(subscriberId)) { + return entry.getKey(); + } + } + return null; + } + + private String getUrlString(DeviceId deviceId, String request) { + RestSBDevice restSBDevice = deviceMap.get(deviceId); + if (restSBDevice == null) { + log.warn("getUrlString::restSbDevice cannot be NULL!"); + return ""; + } + if (restSBDevice.url() != null) { + return restSBDevice.protocol() + COLON + DOUBLESLASH + restSBDevice.url() + request; + } else { + return restSBDevice.protocol() + COLON + DOUBLESLASH + restSBDevice.ip().toString() + + COLON + restSBDevice.port() + request; + } + } + + private String getSubscriptionIdFromDeviceId(DeviceId deviceId) { + if (subscribedDevicesTable.containsKey(deviceId)) { + return subscribedDevicesTable.get(deviceId); + } + return null; + } + + private BlockingQueue<String> getEventQ(DeviceId deviceId) { + if (eventQMap.containsKey(deviceId)) { + return eventQMap.get(deviceId); + } + return null; + } + + /** + * Returns restconfApiCallNode. + * + * @return restconfApiCallNode + */ + protected RestconfApiCallNode restconfapiCallNode() { + return restconfApiCallNode; + } + + /** + * Sets restconfApiCallNode. + * + * @param node restconfApiCallNode + */ + void restconfapiCallNode(RestconfApiCallNode node) { + restconfApiCallNode = node; + } + + Map<String, SubscriptionInfo> subscriptionInfoMap() { + return subscriptionInfoMap; + } + + void subscriptionInfoMap(Map<String, SubscriptionInfo> subscriptionInfoMap) { + this.subscriptionInfoMap = subscriptionInfoMap; + } + + LinkedBlockingQueue<String> eventQueue() { + return eventQueue; + } + + void eventQueue(LinkedBlockingQueue<String> eventQueue) { + this.eventQueue = eventQueue; + } + + /** + * Establishes a persistent SSE connection between the client and the server. + * + * @param paramMap input paramter map + * @param ctx service logic context + */ + @Override + public void establishPersistentSseConnection(Map<String, String> paramMap, SvcLogicContext ctx) throws SvcLogicException { + + } + + @Override + public void establishPeriodicPullConnection(Map<String, String> paramMap, SvcLogicContext ctx) throws SvcLogicException { + String subscriberId = paramMap.get(SUBSCRIBER_ID); + SvcLogicGraphInfo callbackDG = new SvcLogicGraphInfo(paramMap.get("module"), + paramMap.get("rpc"), + paramMap.get("version"), + paramMap.get("mode")); + SubscriptionInfo info = new SubscriptionInfo(); + info.callBackDG(callbackDG); + info.subscriberId(subscriberId); + + String periodicPullUrlString = paramMap.get(PERIODIC_PUL_URL); + URL periodicPullUrl = null; + RestSBDevice dev = null; + try { + periodicPullUrl = new URL(periodicPullUrlString); + dev = getDevice(periodicPullUrl.getHost(), periodicPullUrl.getPort()); + } catch (MalformedURLException e) { + log.error("establishPersistentSseConnection::MalformedURLException happened. e: {}", e); + return; + } + + if (dev == null) { + log.warn("establishPeriodicPullConnection::device does not exist in the map. Trying to add one now."); + dev = new DefaultRestSBDevice(periodicPullUrl.getHost(), + periodicPullUrl.getPort(), "onos", "rocks", "http", + periodicPullUrl.getHost() + ":" + periodicPullUrl.getPort(), true); + this.addDevice(dev); + } + + if (isNotificationEnabled(dev.deviceId())) { + log.warn("establishPeriodicPullConnection::notifications already enabled on device: {}", + dev.deviceId()); + return; + } + + if (getSubscriptionIdFromDeviceId(dev.deviceId()) == null) { + log.warn("This device {} has not yet been subscribed to receive notifications.", + dev.deviceId()); + return; + } + + RestconfNotificationEventListenerImpl myListener = + new RestconfNotificationEventListenerImpl(info); + enableNotifications(dev.deviceId(), "ietf-service-pm:performance-monitoring", "json", myListener); + } + + @Override + public void enableNotifications(DeviceId device, String request, + String mediaType, + RestconfNotificationEventListener listener) { + if (isNotificationEnabled(device)) { + log.warn("enableNotifications::already enabled on device: {}", device); + return; + } + + request = discoverRootResource(device) + RESOURCE_PATH_PREFIX + + request; + + addNotificationListener(device, listener); + + PeriodicPullRunnable periodicRunnable = new PeriodicPullRunnable(request, device); + periodicRunnableTable.put(device, periodicRunnable); + scheduledExecutor.scheduleAtFixedRate(periodicRunnable, 0, 60, TimeUnit.SECONDS); + } + + public void stopNotifications(DeviceId device) { + try { + periodicRunnableTable.get(device).terminate(); + processorRunnableTable.get(device).terminate(); + } catch (Exception ex) { + log.error("stopNotifications::Exception happened when terminating, ex: {}", ex); + } + log.info("stopNotifications::Runnable is now terminated"); + periodicRunnableTable.remove(device); + processorRunnableTable.remove(device); + log.debug("stopNotifications::Stop sending notifications for device URI: " + device.uri().toString()); + } + + @Override + public void deleteSubscriptionAndSseConnection(Map<String, String> paramMap, SvcLogicContext ctx) { + String deleteSubscribeUrlString = paramMap.get(REST_API_URL); + URL deleteSubscribeUrl = null; + RestSBDevice dev = null; + try { + deleteSubscribeUrl = new URL(deleteSubscribeUrlString); + dev = getDevice(deleteSubscribeUrl.getHost(), deleteSubscribeUrl.getPort()); + } catch (MalformedURLException e) { + log.error("establishSubscriptionOnly::MalformedURLException happened. e: {}", e); + return; + } + + String deviceIp = deleteSubscribeUrl.getHost(); + String devicePort = String.valueOf(deleteSubscribeUrl.getPort()); + log.info("deleteSubscriptionAndSseConnection::Trying to unsubscribe device {}:{}", + deviceIp, devicePort); + if (dev == null) { + log.error("deleteSubscriptionAndSseConnection::device does not exist in the map"); + return; + } + String subscriptionId = getSubscriptionIdFromDeviceId(dev.deviceId()); + + if (subscriptionId != null) { + log.info("deleteSubscriptionAndSseConnection::SubscriptionID is found {}", subscriptionId); + log.info("deleteSubscriptionAndSseConnection::About to send unsubscribe request"); + try { + ctx.setAttribute("subscriptionId", subscriptionId); + restapiCallNode.sendRequest(paramMap, ctx); + if (getResponseCode(paramMap.get(RESPONSE_PREFIX), ctx).equals(RESPONSE_CODE_200)) { + log.info("deleteSubscriptionAndSseConnection::Successfully unsubscribed"); + stopNotifications(dev.deviceId()); + subscribedDevicesTable.remove(dev.deviceId()); + + String id = getSubscriptionId(paramMap.get(SUBSCRIBER_ID)); + if (id != null) { + subscriptionInfoMap.remove(id); + } + + } else { + log.info("deleteSubscriptionAndSseConnection::Unsubscription was NOT successfull"); + } + } catch (SvcLogicException e) { + log.error("deleteSubscriptionAndSseConnection::Exception happened ex: {}", e); + } + } else { + log.warn("deleteSubscriptionAndSseConnection::This device has already been unsubscribed"); + } + } + + public class PeriodicPullRunnable implements Runnable { + private String request; + private DeviceId deviceId; + + private volatile boolean running = true; + + public void terminate() { + log.info("PeriodicPullRunnable.terminate()::threadID: {}", + Thread.currentThread().getId()); + running = false; + } + + /** + * @param request request + * @param deviceId device identifier + */ + public PeriodicPullRunnable(String request, DeviceId deviceId) { + this.request = request; + this.deviceId = deviceId; + } + + @Override + public void run() { + log.trace("PeriodicPullRunnable.run()::threadID is: {} ...., running is: {}", + Thread.currentThread().getId(), running); + try { + Client client = ClientBuilder.newBuilder().build(); + WebTarget target = client.target(getUrlString(deviceId, request)); + log.trace("PeriodicPullRunnable.run()::target URI is {}", target.getUri().toString()); + Response response = null; + if (running) { + response = target.request().get(); + String rcvdData = response.readEntity(String.class); + log.trace("PeriodicPullRunnable.run()::after readEntity"); + BlockingQueue<String> eventQ = getEventQ(deviceId); + if (eventQ != null) { + eventQ.add(rcvdData); + eventQMap.put(deviceId, eventQ); + log.trace("PeriodicPullRunnable.run()::eventQ got filled."); + } else { + log.error("PeriodicPullRunnable.run()::eventQ has not been initialized for this device {}", + deviceId); + } + } else { + log.trace("PeriodicPullRunnable.run()::running is false! " + + "closing the client and the response, threadID: {}", Thread.currentThread().getId()); + response.close(); + client.close(); + log.info("PeriodicPullRunnable.run()::eventInput is closed in run()"); + } + } catch (Exception ex) { + log.info("PeriodicPullRunnable.run()::We got some exception: {}, threadID: {} ", ex, + Thread.currentThread().getId()); + } + log.trace("PeriodicPullRunnable.run()::after Runnable Try Catch. threadID: {} ", + Thread.currentThread().getId()); + } + } + + public class InternalPeriodicPullingProcessorRunnable implements Runnable { + + private volatile boolean running = true; + private DeviceId deviceId; + + public InternalPeriodicPullingProcessorRunnable(DeviceId deviceId) { + this.deviceId = deviceId; + } + + public void terminate() { + log.info("InternalPeriodicPullingProcessorRunnable.terminate()::threadID: {}", + Thread.currentThread().getId()); + running = false; + } + + @Override + public void run() { + log.trace("InternalPeriodicPullingProcessorRunnable::restconf event processor runnable inside run()"); + while (running) { + try { + if (eventQMap != null && !eventQMap.isEmpty() && eventQMap.get(deviceId) != null) { + log.trace("InternalPeriodicPullingProcessorRunnable::waiting for take()"); + if (running) { + String eventJsonString = eventQMap.get(deviceId).take(); + log.trace("InternalPeriodicPullingProcessorRunnable::after take()"); + log.info("InternalPeriodicPullingProcessorRunnable::eventJsonString is {}", eventJsonString); + Map<String, String> param = convertToProperties(eventJsonString); + String idString = param.get("push-change-update.subscription-id"); + SubscriptionInfo info = subscriptionInfoMap().get(idString); + if (info != null) { + SvcLogicContext ctx = setContext(param); + SvcLogicGraphInfo callbackDG = info.callBackDG(); + callbackDG.executeGraph(ctx); + } + } else { + log.info("InternalPeriodicPullingProcessorRunnable.run()::running has changed to false " + + "while eventQ was blocked to process new notifications"); + log.info("InternalPeriodicPullingProcessorRunnable.run()::" + + "the client is no longer interested to receive notifications."); + break; + } + } + } catch (InterruptedException | SvcLogicException e) { + e.printStackTrace(); + } + } + } + private SvcLogicContext setContext(Map<String, String> param) { + SvcLogicContext ctx = new SvcLogicContext(); + for (Map.Entry<String, String> entry : param.entrySet()) { + ctx.setAttribute(entry.getKey(), entry.getValue()); + } + return ctx; + } + } + + public String discoverRootResource(DeviceId device) { + return ROOT_RESOURCE; + } + + @Override + public void addNotificationListener(DeviceId deviceId, + RestconfNotificationEventListener listener) { + } + + @Override + public void removeNotificationListener(DeviceId deviceId, + RestconfNotificationEventListener listener) { + } + + public boolean isNotificationEnabled(DeviceId deviceId) { + return periodicRunnableTable.containsKey(deviceId); + } + +} diff --git a/plugins/restconf-client/provider/src/main/java/org/onap/ccsdk/sli/plugins/restconfdiscovery/RestconfDiscoveryNode.java b/plugins/restconf-client/provider/src/main/java/org/onap/ccsdk/sli/plugins/restconfdiscovery/RestconfDiscoveryNode.java index e3b051265..10529b3ab 100644 --- a/plugins/restconf-client/provider/src/main/java/org/onap/ccsdk/sli/plugins/restconfdiscovery/RestconfDiscoveryNode.java +++ b/plugins/restconf-client/provider/src/main/java/org/onap/ccsdk/sli/plugins/restconfdiscovery/RestconfDiscoveryNode.java @@ -562,6 +562,11 @@ public class RestconfDiscoveryNode implements RestConfSBController, SvcLogicDisc } @Override + public void establishPeriodicPullConnection(Map<String, String> paramMap, SvcLogicContext ctx) throws SvcLogicException { + + } + + @Override public void enableNotifications(DeviceId device, String request, String mediaType, RestconfNotificationEventListener listener) { diff --git a/plugins/restconf-client/provider/src/main/java/org/onap/ccsdk/sli/plugins/restconfdiscovery/SvcLogicDiscoveryPlugin.java b/plugins/restconf-client/provider/src/main/java/org/onap/ccsdk/sli/plugins/restconfdiscovery/SvcLogicDiscoveryPlugin.java index 972fb2b55..d40d9a3f8 100644 --- a/plugins/restconf-client/provider/src/main/java/org/onap/ccsdk/sli/plugins/restconfdiscovery/SvcLogicDiscoveryPlugin.java +++ b/plugins/restconf-client/provider/src/main/java/org/onap/ccsdk/sli/plugins/restconfdiscovery/SvcLogicDiscoveryPlugin.java @@ -124,6 +124,36 @@ public interface SvcLogicDiscoveryPlugin extends SvcLogicJavaPlugin { void establishPersistentSseConnection(Map<String, String> paramMap, SvcLogicContext ctx) throws SvcLogicException; /** + * Allows directed graphs to establish a periodic pull from a given controller. + * @param paramMap HashMap<String,String> of parameters passed by the DG to this function + * <table border="1"> + * <thead><th>parameter</th><th>Mandatory/Optional</th><th>description</th><th>example values</th></thead> + * <tbody> + * <tr><td>templateDirName</td><td>Optional</td><td>full path to YANG directory that can be used to build a request</td><td>/sdncopt/bvc/resconfapi/test</td></tr> + * <tr><td>periodicPullURL</td><td>Mandatory</td><td>url to send periodically to the server</td><td>https://127.0.0.1:8181/restconf/data/ietf-service-pm:performance-monitoring</td></tr> + * <tr><td>callbackDG</td><td>Mandatory</td><td>callback DG to process the received notification</td><td>Resource-Discovery:handleSOTNTopology</td></tr> + * <tr><td>filterURL</td><td>Optional</td><td>url which needs to be subscribed, if null subscribe to all</td><td>http://example.com/sample-data/1.0</td></tr> + * <tr><td>subscriptionType</td><td>Optional</td><td>type of subscription, periodic or onDataChange</td><td>onDataChange</td></tr> + * <tr><td>updateFrequency</td><td>Optional</td><td>update frequency in milli seconds when subscription type is periodic</td><td>1000</td></tr> + * <tr><td>restapiUser</td><td>Optional</td><td>user name to use for http basic authentication</td><td>sdnc_ws</td></tr> + * <tr><td>restapiPassword</td><td>Optional</td><td>unencrypted password to use for http basic authentication</td><td>plain_password</td></tr> + * <tr><td>contentType</td><td>Optional</td><td>http content type to set in the http header</td><td>usually application/json or application/xml</td></tr> + * <tr><td>format</td><td>Optional</td><td>should match request body format</td><td>json or xml</td></tr> + * <tr><td>responsePrefix</td><td>Optional</td><td>location the notification response will be written to in context memory</td><td>tmp.restconfdiscovery.result</td></tr> + * <tr><td>skipSending</td><td>Optional</td><td></td><td>true or false</td></tr> + * <tr><td>convertResponse </td><td>Optional</td><td>whether the response should be converted</td><td>true or false</td></tr> + * <tr><td>customHttpHeaders</td><td>Optional</td><td>a list additional http headers to be passed in, follow the format in the example</td><td>X-CSI-MessageId=messageId,headerFieldName=headerFieldValue</td></tr> + * <tr><td>dumpHeaders</td><td>Optional</td><td>when true writes http header content to context memory</td><td>true or false</td></tr> + * </tbody> + * </table> + * @param ctx Reference to context memory + * @throws SvcLogicException + * @since 11.0.2 + * @see String#split(String, int) + */ + void establishPeriodicPullConnection(Map<String, String> paramMap, SvcLogicContext ctx) throws SvcLogicException; + + /** * Allows directed graphs to modify a discovery subscription for a given subscriber. * @param paramMap HashMap<String,String> of parameters passed by the DG to this function * <table border="1"> diff --git a/plugins/restconf-client/provider/src/main/resources/org/opendaylight/blueprint/restconf-client-blueprint.xml b/plugins/restconf-client/provider/src/main/resources/org/opendaylight/blueprint/restconf-client-blueprint.xml index 20a3ab2b5..3e4e1fe06 100755 --- a/plugins/restconf-client/provider/src/main/resources/org/opendaylight/blueprint/restconf-client-blueprint.xml +++ b/plugins/restconf-client/provider/src/main/resources/org/opendaylight/blueprint/restconf-client-blueprint.xml @@ -43,6 +43,10 @@ <argument ref="restconfapiCallNodeProvider"/> </bean> + <bean id="periodicDiscoveryNodeProvider" class="org.onap.ccsdk.sli.plugins.restconfdiscovery.PeriodicDiscoveryNode" > + <argument ref="restconfapiCallNodeProvider"/> + </bean> + <service ref="restconfapiCallNodeProvider"> <interfaces> <value>org.onap.ccsdk.sli.plugins.restconfapicall.RestconfApiCallNode</value> @@ -55,4 +59,10 @@ </interfaces> </service> + <service ref="periodicDiscoveryNodeProvider"> + <interfaces> + <value>org.onap.ccsdk.sli.plugins.restconfdiscovery.PeriodicDiscoveryNode</value> + </interfaces> + </service> + </blueprint>
\ No newline at end of file |