ONAP-PDP-REST/src/main/java/org/onap/policy/pdp/rest/notifications/NotificationServer.java

   1 /*-
   2  * ============LICENSE_START=======================================================
   3  * ONAP-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.onap.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.onap.policy.common.logging.eelf.MessageCodes;
  42 import org.onap.policy.common.logging.eelf.PolicyLogger;
  43 import org.onap.policy.common.logging.flexlogger.FlexLogger;
  44 import org.onap.policy.common.logging.flexlogger.Logger;
  45 import org.onap.policy.pdp.rest.api.services.NotificationService;
  46 import org.onap.policy.rest.XACMLRestProperties;
  47 import org.onap.policy.utils.BusPublisher;
  48 import org.onap.policy.xacml.api.XACMLErrorConstants;
  49
  50 import com.att.nsa.cambria.client.CambriaBatchingPublisher;
  51 import com.att.nsa.cambria.client.CambriaClientBuilders;
  52 import com.att.research.xacml.util.XACMLProperties;
  53
  54
  55 /**
  56  * The NotificationServer sends the Server Notifications to the Clients once there is any Event.
  57  * WebSockets is being used as a medium for sending Notifications.
  58  * UEB is being used as a medium for sending Notifications.
  59  * DMAAP is being used as a medium for sending Notifications.
  60  *
  61  * @version 0.2
  62  *
  63  **/
  64 @ServerEndpoint(value = "/notifications")
  65 public class NotificationServer {
  66         private static final Logger LOGGER      = FlexLogger.getLogger(NotificationServer.class);
  67         private static Queue<Session> queue = new ConcurrentLinkedQueue<>();
  68         private static String update = null;
  69         private static  String hosts = null;
  70         private static URL aURL = null;
  71
  72         @OnOpen
  73         public void openConnection(Session session) {
  74                 LOGGER.info("Session Connected: " + session.getId());
  75                 queue.add(session);
  76         }
  77
  78         @OnClose
  79         public void closeConnection(Session session) {
  80                 queue.remove(session);
  81         }
  82
  83         @OnError
  84         public void error(Session session, Throwable t) {
  85                 queue.remove(session);
  86                 LOGGER.info(XACMLErrorConstants.ERROR_PROCESS_FLOW + "Session Error for : " + session.getId() + " Error: " + t.getMessage());
  87
  88         }
  89
  90         @OnMessage
  91         public void Message(String message, Session session) {
  92
  93                 if(message.equalsIgnoreCase("Manual")) {
  94                         try {
  95                                 session.getBasicRemote().sendText(update);
  96                                 session.close();
  97                         } catch (IOException e) {
  98                                 LOGGER.info(XACMLErrorConstants.ERROR_PROCESS_FLOW + "Error in sending the Event Notification: "+ e.getMessage() + e);
  99                                 LOGGER.error(XACMLErrorConstants.ERROR_PROCESS_FLOW + "Error sending Message update");
 100                         }
 101                 }
 102         }
 103
 104         public static void sendNotification(String notification, String propNotificationType, String pdpURL) throws Exception {
 105
 106                 LOGGER.debug("Notification set to " + propNotificationType);
 107                 if (propNotificationType.equals("ueb")){
 108
 109                         String topic = null;
 110                         try {
 111                                 aURL = new URL(pdpURL);
 112                                 topic = aURL.getHost() + aURL.getPort();
 113                         } catch (MalformedURLException e1) {
 114                                 pdpURL = pdpURL.replace("/", "");
 115                                 topic = pdpURL.replace(":", "");
 116                                 LOGGER.error(XACMLErrorConstants.ERROR_PROCESS_FLOW + "Error in parsing out pdpURL for UEB notfication ");
 117                                 PolicyLogger.error(MessageCodes.ERROR_PROCESS_FLOW, e1, "Error in parsing out pdpURL for UEB notfication ");
 118                         }
 119                         hosts = XACMLProperties.getProperty(XACMLRestProperties.PROP_NOTIFICATION_SERVERS);
 120                         String apiKey = XACMLProperties.getProperty(XACMLRestProperties.PROP_UEB_API_KEY);
 121                         String apiSecret = XACMLProperties.getProperty(XACMLRestProperties.PROP_UEB_API_SECRET);
 122
 123                         LOGGER.debug("Creating Publisher for host: " + hosts + " with topic: " + topic);
 124                         CambriaBatchingPublisher pub = null;
 125                         try {
 126                                 if(hosts==null || topic==null || apiKey==null || apiSecret==null){
 127                                         LOGGER.error(XACMLErrorConstants.ERROR_DATA_ISSUE + "UEB properties are missing from the property file ");
 128                                         throw new Exception(XACMLErrorConstants.ERROR_DATA_ISSUE + "UEB properties are missing from the property file ");
 129                                 }
 130
 131                                 hosts.trim();
 132                                 topic.trim();
 133                                 apiKey.trim();
 134                                 apiSecret.trim();
 135                                 pub = new CambriaClientBuilders.PublisherBuilder ()
 136                                                 .usingHosts ( hosts )
 137                                                 .onTopic ( topic )
 138                                                 .authenticatedBy ( apiKey, apiSecret )
 139                                                 .build ()
 140                                                 ;
 141
 142                         } catch (MalformedURLException e1) {
 143                                 LOGGER.error(XACMLErrorConstants.ERROR_PROCESS_FLOW + "Error creating the UEB publisher" + e1.getMessage());
 144                         } catch (GeneralSecurityException e1) {
 145                                 LOGGER.error(XACMLErrorConstants.ERROR_PROCESS_FLOW + "Error creating the UEB publisher" + e1.getMessage() +e1);
 146                         }
 147                         if(pub != null){
 148                                 try {
 149                                         pub.send( "MyPartitionKey", notification );
 150                                 } catch (IOException e) {
 151                                         LOGGER.error(XACMLErrorConstants.ERROR_PROCESS_FLOW + "Error sending notification update" + e.getMessage() + e);
 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.isEmpty()){
 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<>(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() + e);
 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() + e);
 215                         }
 216                 }
 217                 NotificationService.sendNotification(notification);
 218         }
 219
 220         public static void setUpdate(String update) {
 221                 NotificationServer.update = update;
 222         }
 223
 224 }