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.ArrayList;
  28 import java.util.Arrays;
  29 import java.util.List;
  30 import java.util.Queue;
  31 import java.util.concurrent.ConcurrentLinkedQueue;
  32 import java.util.concurrent.TimeUnit;
  33
  34 import javax.websocket.OnClose;
  35 import javax.websocket.OnError;
  36 import javax.websocket.OnMessage;
  37 import javax.websocket.OnOpen;
  38 import javax.websocket.Session;
  39 import javax.websocket.server.ServerEndpoint;
  40
  41 import org.openecomp.policy.common.logging.eelf.MessageCodes;
  42 import org.openecomp.policy.common.logging.eelf.PolicyLogger;
  43 import org.openecomp.policy.common.logging.flexlogger.FlexLogger;
  44 import org.openecomp.policy.common.logging.flexlogger.Logger;
  45 import org.openecomp.policy.rest.XACMLRestProperties;
  46 import org.openecomp.policy.utils.BusPublisher;
  47 import org.openecomp.policy.xacml.api.XACMLErrorConstants;
  48
  49 import com.att.nsa.cambria.client.CambriaBatchingPublisher;
  50 import com.att.nsa.cambria.client.CambriaClientBuilders;
  51 import com.att.research.xacml.util.XACMLProperties;
  52
  53
  54 /**
  55  * The NotificationServer sends the Server Notifications to the Clients once there is any Event.
  56  * WebSockets is being used as a medium for sending Notifications.
  57  * UEB is being used as a medium for sending Notifications.
  58  * DMAAP is being used as a medium for sending Notifications.
  59  *
  60  * @version 0.2
  61  *
  62  **/
  63 @ServerEndpoint(value = "/notifications")
  64 public class NotificationServer {
  65         private static final Logger LOGGER      = FlexLogger.getLogger(NotificationServer.class);
  66         private static Queue<Session> queue = new ConcurrentLinkedQueue<Session>();
  67         private static String update = null;
  68         private static  String hosts = null;
  69         private static URL aURL = null;
  70
  71         @OnOpen
  72         public void openConnection(Session session) {
  73                 LOGGER.info("Session Connected: " + session.getId());
  74                 queue.add(session);
  75         }
  76
  77         @OnClose
  78         public void closeConnection(Session session) {
  79                 queue.remove(session);
  80         }
  81
  82         @OnError
  83         public void error(Session session, Throwable t) {
  84                 queue.remove(session);
  85                 LOGGER.info(XACMLErrorConstants.ERROR_PROCESS_FLOW + "Session Error for : " + session.getId() + " Error: " + t.getMessage());
  86
  87         }
  88
  89         @OnMessage
  90         public void Message(String message, Session session) {
  91
  92                 if(message.equalsIgnoreCase("Manual")) {
  93                         try {
  94                                 session.getBasicRemote().sendText(update);
  95                                 session.close();
  96                         } catch (IOException e) {
  97                                 LOGGER.info(XACMLErrorConstants.ERROR_PROCESS_FLOW + "Error in sending the Event Notification: "+ e.getMessage());
  98                                 LOGGER.error(XACMLErrorConstants.ERROR_PROCESS_FLOW + "Error sending Message update");
  99                         }
 100                 }
 101         }
 102
 103         public static void sendNotification(String notification, String propNotificationType, String pdpURL) throws Exception {
 104
 105                 LOGGER.debug("Notification set to " + propNotificationType);
 106                 if (propNotificationType.equals("ueb")){
 107
 108                         String topic = null;
 109                         try {
 110                                 aURL = new URL(pdpURL);
 111                                 topic = aURL.getHost() + aURL.getPort();
 112                         } catch (MalformedURLException e1) {
 113                                 pdpURL = pdpURL.replace("/", "");
 114                                 topic = pdpURL.replace(":", "");
 115                                 LOGGER.error(XACMLErrorConstants.ERROR_PROCESS_FLOW + "Error in parsing out pdpURL for UEB notfication ");
 116                                 PolicyLogger.error(MessageCodes.ERROR_PROCESS_FLOW, e1, "Error in parsing out pdpURL for UEB notfication ");
 117                         }
 118                         hosts = XACMLProperties.getProperty(XACMLRestProperties.PROP_NOTIFICATION_SERVERS);
 119                         String apiKey = XACMLProperties.getProperty(XACMLRestProperties.PROP_UEB_API_KEY);
 120                         String apiSecret = XACMLProperties.getProperty(XACMLRestProperties.PROP_UEB_API_SECRET);
 121
 122                         LOGGER.debug("Creating Publisher for host: " + hosts + " with topic: " + topic);
 123                         CambriaBatchingPublisher pub = null;
 124                         try {
 125                                 if(hosts==null || topic==null || apiKey==null || apiSecret==null){
 126                                         LOGGER.error(XACMLErrorConstants.ERROR_DATA_ISSUE + "UEB properties are missing from the property file ");
 127                                         throw new Exception(XACMLErrorConstants.ERROR_DATA_ISSUE + "UEB properties are missing from the property file ");
 128                                 }
 129
 130                                 hosts.trim();
 131                                 topic.trim();
 132                                 apiKey.trim();
 133                                 apiSecret.trim();
 134                                 pub = new CambriaClientBuilders.PublisherBuilder ()
 135                                                 .usingHosts ( hosts )
 136                                                 .onTopic ( topic )
 137                                                 .authenticatedBy ( apiKey, apiSecret )
 138                                                 .build ()
 139                                         ;
 140
 141                         } catch (MalformedURLException e1) {
 142                                 LOGGER.error(XACMLErrorConstants.ERROR_PROCESS_FLOW + "Error creating the UEB publisher" + e1.getMessage());
 143                         } catch (GeneralSecurityException e1) {
 144                                 LOGGER.error(XACMLErrorConstants.ERROR_PROCESS_FLOW + "Error creating the UEB publisher" + e1.getMessage());
 145                         }
 146
 147                         try {
 148                                 pub.send( "MyPartitionKey", notification );
 149                         } catch (IOException e) {
 150                                 LOGGER.error(XACMLErrorConstants.ERROR_PROCESS_FLOW + "Error sending notification update" + e.getMessage());
 151                         }
 152
 153                         // close the publisher. The batching publisher does not send events
 154                         // immediately, so you MUST use close to send any remaining messages.
 155                         // You provide the amount of time you're willing to wait for the sends
 156                         // to succeed before giving up. If any messages are unsent after that time,
 157                         // they're returned to your app. You could, for example, persist to disk
 158                         // and try again later.
 159                         final List<?> stuck = pub.close ( 20, TimeUnit.SECONDS );
 160
 161                         if ( stuck.size () > 0 ){
 162                                 LOGGER.error( stuck.size() + " messages unsent" );
 163                         }else{
 164                                 LOGGER.info( "Clean exit; all messages sent: " + notification );
 165                         }
 166
 167                 } else if (propNotificationType.equals("dmaap")) {
 168
 169                         // Setting up the Publisher for DMaaP MR
 170                         String dmaapServers = XACMLProperties.getProperty(XACMLRestProperties.PROP_NOTIFICATION_SERVERS);
 171                         String topic = XACMLProperties.getProperty(XACMLRestProperties.PROP_NOTIFICATION_TOPIC);
 172                         String aafLogin = XACMLProperties.getProperty("DMAAP_AAF_LOGIN");
 173                         String aafPassword = XACMLProperties.getProperty("DMAAP_AAF_PASSWORD");
 174
 175                         try {
 176                                 if(dmaapServers==null || topic==null){
 177                                         LOGGER.error(XACMLErrorConstants.ERROR_DATA_ISSUE + "DMaaP properties are missing from the property file ");
 178                                         throw new Exception(XACMLErrorConstants.ERROR_DATA_ISSUE + "DMaaP properties are missing from the property file ");
 179                                 }
 180
 181                                 dmaapServers.trim();
 182                                 topic.trim();
 183                                 aafLogin.trim();
 184                                 aafPassword.trim();
 185
 186                                 List<String> dmaapList = null;
 187                                 if(dmaapServers.contains(",")) {
 188                                         dmaapList = new ArrayList<String>(Arrays.asList(dmaapServers.split("\\s*,\\s*")));
 189                                 } else {
 190                                         dmaapList = new ArrayList<>();
 191                                         dmaapList.add(dmaapServers);
 192                                 }
 193
 194                                 BusPublisher publisher =
 195                                                 new BusPublisher.DmaapPublisherWrapper(dmaapList,
 196                                                                                                topic,
 197                                                                                                aafLogin,
 198                                                                                                aafPassword);
 199
 200                                 // Sending notification through DMaaP Message Router
 201                                 publisher.send( "MyPartitionKey", notification);
 202                                 LOGGER.debug("Message Published on DMaaP :" + dmaapList.get(0) + "for Topic: " + topic);
 203                                 publisher.close();
 204
 205                         } catch (Exception e) {
 206                                 LOGGER.error(XACMLErrorConstants.ERROR_PROCESS_FLOW + "Error sending notification update" + e.getMessage());
 207                         }
 208                 }
 209
 210                 for(Session session: queue) {
 211                         try {
 212                                 session.getBasicRemote().sendText(notification);
 213                         } catch (IOException e) {
 214                                 LOGGER.info(XACMLErrorConstants.ERROR_PROCESS_FLOW + "Error in sending the Event Notification: "+ e.getMessage());
 215                         }
 216                 }
 217         }
 218
 219         public static void setUpdate(String update) {
 220                 NotificationServer.update = update;
 221         }
 222 }