2 * ============LICENSE_START=======================================================
4 * ================================================================================
5 * Copyright (C) 2018 Huawei Technologies Co., Ltd. All rights reserved.
6 * ================================================================================
7 * Licensed under the Apache License, Version 2.0 (the "License");
8 * you may not use this file except in compliance with the License.
9 * You may obtain a copy of the License at
11 * http://www.apache.org/licenses/LICENSE-2.0
13 * Unless required by applicable law or agreed to in writing, software
14 * distributed under the License is distributed on an "AS IS" BASIS,
15 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
16 * See the License for the specific language governing permissions and
17 * limitations under the License.
18 * ============LICENSE_END=========================================================
21 package org.onap.ccsdk.sli.plugins.restconfdiscovery;
23 import org.glassfish.jersey.media.sse.EventSource;
24 import org.glassfish.jersey.media.sse.SseFeature;
25 import org.onap.ccsdk.sli.core.sli.SvcLogicContext;
26 import org.onap.ccsdk.sli.core.sli.SvcLogicException;
27 import org.onap.ccsdk.sli.plugins.restapicall.Parameters;
28 import org.onap.ccsdk.sli.plugins.restapicall.RestapiCallNode;
29 import org.onap.ccsdk.sli.plugins.restconfapicall.RestconfApiCallNode;
30 import org.slf4j.Logger;
32 import javax.net.ssl.SSLContext;
33 import javax.net.ssl.TrustManager;
34 import javax.net.ssl.X509TrustManager;
35 import javax.ws.rs.client.Client;
36 import javax.ws.rs.client.ClientBuilder;
37 import javax.ws.rs.client.WebTarget;
38 import java.security.KeyManagementException;
39 import java.security.NoSuchAlgorithmException;
40 import java.security.cert.CertificateException;
41 import java.security.cert.X509Certificate;
43 import java.util.concurrent.ConcurrentHashMap;
44 import java.util.concurrent.ExecutorService;
45 import java.util.concurrent.Executors;
46 import java.util.concurrent.LinkedBlockingQueue;
48 import static org.slf4j.LoggerFactory.getLogger;
51 * Representation of a plugin to subscribe for notification and then
52 * to handle the received notifications.
54 public class RestconfDiscoveryNode implements SvcLogicDiscoveryPlugin {
56 private static final Logger log = getLogger(RestconfDiscoveryNode.class);
58 private ExecutorService executor = Executors.newCachedThreadPool();
59 private Map<String, PersistentConnection> runnableInfo = new ConcurrentHashMap<>();
60 private RestconfApiCallNode restconfApiCallNode;
62 private volatile Map<String, SubscriptionInfo> subscriptionInfoMap = new ConcurrentHashMap<>();
63 private volatile LinkedBlockingQueue<String> eventQueue = new LinkedBlockingQueue<>();
65 private static final String SUBSCRIBER_ID = "subscriberId";
66 private static final String RESPONSE_CODE = "response-code";
67 private static final String RESPONSE_PREFIX = "responsePrefix";
68 private static final String OUTPUT_IDENTIFIER = "ietf-subscribed-notif" +
69 "ications:establish-subscription.output.identifier";
70 private static final String RESPONSE_CODE_200 = "200";
71 private static final String SSE_URL = "sseConnectURL";
74 * Creates an instance of RestconfDiscoveryNode and starts processing of
77 * @param r restconf api call node
79 public RestconfDiscoveryNode(RestconfApiCallNode r) {
80 this.restconfApiCallNode = r;
81 ExecutorService e = Executors.newFixedThreadPool(20);
82 EventProcessor p = new EventProcessor(this);
83 for (int i = 0; i < 20; ++i) {
89 public void establishSubscription(Map<String, String> paramMap,
90 SvcLogicContext ctx) throws SvcLogicException {
91 String subscriberId = paramMap.get(SUBSCRIBER_ID);
92 if (subscriberId == null) {
93 throw new SvcLogicException("Subscriber Id is null");
96 restconfApiCallNode.sendRequest(paramMap, ctx);
98 if (getResponseCode(paramMap.get(RESPONSE_PREFIX), ctx).equals(RESPONSE_CODE_200)) {
99 // TODO: save subscription id and subscriber in MYSQL
101 establishPersistentConnection(paramMap, ctx, subscriberId);
103 log.info("Failed to subscribe {}", subscriberId);
104 throw new SvcLogicException(ctx.getAttribute(RESPONSE_CODE));
109 public void modifySubscription(Map<String, String> paramMap, SvcLogicContext ctx) {
110 // TODO: to be implemented
114 public void deleteSubscription(Map<String, String> paramMap, SvcLogicContext ctx) {
115 String id = getSubscriptionId(paramMap.get(SUBSCRIBER_ID));
117 PersistentConnection conn = runnableInfo.get(id);
119 runnableInfo.remove(id);
120 subscriptionInfoMap.remove(id);
124 class PersistentConnection implements Runnable {
126 private volatile boolean running = true;
127 private Map<String, String> paramMap;
129 PersistentConnection(String url, Map<String, String> paramMap) {
131 this.paramMap = paramMap;
134 private void terminate() {
141 WebTarget target = null;
143 RestapiCallNode restapi = restconfApiCallNode.getRestapiCallNode();
144 p = restapi.getParameters(paramMap, new Parameters());
145 Client client = ignoreSslClient().register(SseFeature.class);
146 target = restapi.addAuthType(client, p).target(url);
147 } catch (SvcLogicException e) {
148 log.error("Exception occured!", e);
149 Thread.currentThread().interrupt();
152 target = addToken(target, paramMap.get("customHttpHeaders"));
153 EventSource eventSource = EventSource.target(target).build();
154 eventSource.register(new EventHandler(RestconfDiscoveryNode.this));
156 log.info("Connected to SSE source");
159 log.info("SSE state " + eventSource.isOpen());
161 } catch (InterruptedException e) {
162 log.error("Interrupted!", e);
163 Thread.currentThread().interrupt();
167 log.info("Closed connection to SSE source");
171 private Client ignoreSslClient() {
172 SSLContext sslcontext = null;
175 sslcontext = SSLContext.getInstance("TLS");
176 sslcontext.init(null, new TrustManager[]{new X509TrustManager() {
178 public void checkClientTrusted(X509Certificate[] arg0, String arg1) throws CertificateException {
182 public void checkServerTrusted(X509Certificate[] arg0, String arg1) throws CertificateException {
186 public X509Certificate[] getAcceptedIssuers() {
187 return new X509Certificate[0];
189 } }, new java.security.SecureRandom());
190 } catch (NoSuchAlgorithmException | KeyManagementException e) {
191 throw new IllegalStateException(e);
194 return ClientBuilder.newBuilder().sslContext(sslcontext).hostnameVerifier((s1, s2) -> true).build();
197 protected String getTokenId(String customHttpHeaders) {
198 if (customHttpHeaders.contains("=")) {
199 String s[] = customHttpHeaders.split("=");
202 return customHttpHeaders;
205 protected WebTarget addToken(WebTarget target, String customHttpHeaders) {
206 if (customHttpHeaders == null) {
210 return new AdditionalHeaderWebTarget(
211 target, getTokenId(customHttpHeaders));
215 * Establishes a persistent between the client and server.
217 * @param paramMap input paramter map
218 * @param ctx service logic context
219 * @param subscriberId subscriber identifier
221 void establishPersistentConnection(Map<String, String> paramMap, SvcLogicContext ctx,
222 String subscriberId) {
223 String id = getOutputIdentifier(paramMap.get(RESPONSE_PREFIX), ctx);
224 SvcLogicGraphInfo callbackDG = new SvcLogicGraphInfo(paramMap.get("module"),
226 paramMap.get("version"),
227 paramMap.get("mode"));
228 SubscriptionInfo info = new SubscriptionInfo();
229 info.callBackDG(callbackDG);
230 info.subscriptionId(id);
231 info.subscriberId(subscriberId);
232 subscriptionInfoMap.put(id, info);
234 String url = paramMap.get(SSE_URL);
235 PersistentConnection connection = new PersistentConnection(url, paramMap);
236 runnableInfo.put(id, connection);
237 executor.execute(connection);
241 * Returns response code.
243 * @param prefix prefix given in input parameter
244 * @param ctx service logic context
245 * @return response code
247 String getResponseCode(String prefix, SvcLogicContext ctx) {
248 return ctx.getAttribute(getPrefix(prefix) + RESPONSE_CODE);
252 * Returns subscription id from event.
254 * @param prefix prefix given in input parameter
255 * @param ctx service logic context
256 * @return subscription id from event
258 String getOutputIdentifier(String prefix, SvcLogicContext ctx) {
259 return ctx.getAttribute(getPrefix(prefix) + OUTPUT_IDENTIFIER);
262 private String getPrefix(String prefix) {
263 return prefix != null ? prefix + "." : "";
266 private String getSubscriptionId(String subscriberId) {
267 for (Map.Entry<String,SubscriptionInfo> entry
268 : subscriptionInfoMap.entrySet()) {
269 if (entry.getValue().subscriberId()
270 .equals(subscriberId)) {
271 return entry.getKey();
278 * Returns restconfApiCallNode.
280 * @return restconfApiCallNode
282 protected RestconfApiCallNode restconfapiCallNode() {
283 return restconfApiCallNode;
287 * Sets restconfApiCallNode.
289 * @param node restconfApiCallNode
291 void restconfapiCallNode(RestconfApiCallNode node) {
292 restconfApiCallNode = node;
295 Map<String, SubscriptionInfo> subscriptionInfoMap() {
296 return subscriptionInfoMap;
299 void subscriptionInfoMap(Map<String, SubscriptionInfo> subscriptionInfoMap) {
300 this.subscriptionInfoMap = subscriptionInfoMap;
303 LinkedBlockingQueue<String> eventQueue() {
307 void eventQueue(LinkedBlockingQueue<String> eventQueue) {
308 this.eventQueue = eventQueue;