ECOMP-PDP-REST/src/main/java/org/openecomp/policy/pdp/rest/notifications/NotificationServer.java

   1 /*-
   2  * ============LICENSE_START=======================================================
   3  * ECOMP-PDP-REST
   4  * ================================================================================
   5  * Copyright (C) 2017 AT&T Intellectual Property. 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
  10  *
  11  *      http://www.apache.org/licenses/LICENSE-2.0
  12  *
  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=========================================================
  19  */
  20
  21 package org.openecomp.policy.pdp.rest.notifications;
  22
  23 import java.io.IOException;
  24 import java.net.MalformedURLException;
  25 import java.net.URL;
  26 import java.security.GeneralSecurityException;
  27 import java.util.Queue;
  28 import java.util.concurrent.ConcurrentLinkedQueue;
  29
  30 import javax.websocket.OnClose;
  31 import javax.websocket.OnError;
  32 import javax.websocket.OnMessage;
  33 import javax.websocket.OnOpen;
  34 import javax.websocket.Session;
  35 import javax.websocket.server.ServerEndpoint;
  36
  37 import org.openecomp.policy.rest.XACMLRestProperties;
  38 import org.openecomp.policy.common.logging.eelf.MessageCodes;
  39 import org.openecomp.policy.common.logging.eelf.PolicyLogger;
  40 import com.att.nsa.cambria.client.CambriaClientFactory;
  41 import com.att.nsa.cambria.client.CambriaPublisher;
  42 import org.openecomp.policy.xacml.api.XACMLErrorConstants;
  43 import com.att.research.xacml.util.XACMLProperties;
  44
  45 import org.openecomp.policy.common.logging.flexlogger.*;
  46
  47
  48 /**
  49  * The NotificationServer sends the Server Notifications to the Clients once there is any Event.
  50  * WebSockets is being used as a medium for sending Notifications.
  51  * UEB is being used as a medium for sending Notifications.
  52  *
  53  * @version 0.1
  54  *
  55  **/
  56 @ServerEndpoint(value = "/notifications")
  57 public class NotificationServer {
  58         private static final Logger logger      = FlexLogger.getLogger(NotificationServer.class);
  59         private static Queue<Session> queue = new ConcurrentLinkedQueue<Session>();
  60         private static String update = null;
  61         private static  String hosts = null;
  62         private static URL aURL = null;
  63
  64         @OnOpen
  65         public void openConnection(Session session) {
  66                 logger.info("Session Connected: " + session.getId());
  67                 queue.add(session);
  68         }
  69
  70         @OnClose
  71         public void closeConnection(Session session) {
  72                 queue.remove(session);
  73         }
  74
  75         @OnError
  76         public void error(Session session, Throwable t) {
  77                 queue.remove(session);
  78                 logger.info(XACMLErrorConstants.ERROR_PROCESS_FLOW + "Session Error for : " + session.getId() + " Error: " + t.getMessage());
  79
  80         }
  81
  82         @OnMessage
  83         public void Message(String message, Session session) {
  84
  85                 if(message.equalsIgnoreCase("Manual")) {
  86                         try {
  87                                 session.getBasicRemote().sendText(update);
  88                                 session.close();
  89                         } catch (IOException e) {
  90                                 logger.info(XACMLErrorConstants.ERROR_PROCESS_FLOW + "Error in sending the Event Notification: "+ e.getMessage());
  91                                 logger.error(XACMLErrorConstants.ERROR_PROCESS_FLOW + "Error sending Message update");
  92                         }
  93                 }
  94         }
  95
  96         public static void sendNotification(String notification, String propNotificationType, String pdpURL){
  97
  98                 logger.debug("Notification set to " + propNotificationType);
  99                 if (propNotificationType.equals("ueb")){
 100                         String topic = null;
 101                         try {
 102                                 aURL = new URL(pdpURL);
 103                                 topic = aURL.getHost() + aURL.getPort();
 104                         } catch (MalformedURLException e1) {
 105                                 pdpURL = pdpURL.replace("/", "");
 106                                 topic = pdpURL.replace(":", "");
 107                                 logger.error(XACMLErrorConstants.ERROR_PROCESS_FLOW + "Error in parsing out pdpURL for UEB notfication ");
 108                                 PolicyLogger.error(MessageCodes.ERROR_PROCESS_FLOW, e1, "Error in parsing out pdpURL for UEB notfication ");
 109                         }
 110                         hosts = XACMLProperties.getProperty(XACMLRestProperties.PROP_NOTIFICATION_UEB_CLUSTER);
 111                         logger.debug("Creating Publisher for host: " + hosts + " with topic: " + topic);
 112                         CambriaPublisher pub = null;
 113                         try {
 114                                 pub = CambriaClientFactory.createSimplePublisher (null, hosts, topic );
 115                         } catch (MalformedURLException e1) {
 116                                 // TODO Auto-generated catch block
 117                                 e1.printStackTrace();
 118                         } catch (GeneralSecurityException e1) {
 119                                 // TODO Auto-generated catch block
 120                                 e1.printStackTrace();
 121                         }
 122                         try {
 123                                 pub.send( "MyPartitionKey", notification );
 124                         } catch (IOException e) {
 125                                 logger.error(XACMLErrorConstants.ERROR_PROCESS_FLOW + "Error sending notification update");
 126                         }
 127                         pub.close();
 128                 }
 129                 for(Session session: queue) {
 130                         try {
 131                                 session.getBasicRemote().sendText(notification);
 132                         } catch (IOException e) {
 133                                 logger.info(XACMLErrorConstants.ERROR_PROCESS_FLOW + "Error in sending the Event Notification: "+ e.getMessage());
 134                         }
 135                 }
 136         }
 137
 138         public static void setUpdate(String update) {
 139                 NotificationServer.update = update;
 140         }
 141 }