PolicyEngineAPI/src/main/java/org/openecomp/policy/std/AutoClientDMAAP.java

   1 package org.openecomp.policy.std;
   2
   3 import java.util.List;
   4 import java.util.UUID;
   5
   6 import org.openecomp.policy.api.NotificationHandler;
   7 import org.openecomp.policy.api.NotificationScheme;
   8 import org.openecomp.policy.api.NotificationType;
   9 import org.openecomp.policy.api.PDPNotification;
  10 import org.openecomp.policy.common.logging.flexlogger.FlexLogger;
  11 import org.openecomp.policy.common.logging.flexlogger.Logger;
  12 import org.openecomp.policy.utils.BusConsumer;
  13 import org.openecomp.policy.xacml.api.XACMLErrorConstants;
  14
  15 public class AutoClientDMAAP implements Runnable {
  16         private static StdPDPNotification notification = null;
  17         private static NotificationScheme scheme = null;
  18         private static NotificationHandler handler = null;
  19         private static String topic = null;
  20         private static boolean status = false;
  21         private static Logger logger = FlexLogger.getLogger(AutoClientDMAAP.class.getName());
  22         private static String notficatioinType = null;
  23         private static BusConsumer dmaapConsumer = null;
  24         private static List<String> dmaapList = null;
  25         private static String aafLogin = null;
  26         private static String aafPassword = null;
  27         public volatile boolean isRunning = false;
  28
  29
  30         public AutoClientDMAAP(List<String> dmaapList, String topic, String aafLogin, String aafPassword) {
  31                AutoClientDMAAP.topic = topic;
  32                AutoClientDMAAP.dmaapList = dmaapList;
  33                AutoClientDMAAP.aafLogin = aafLogin;
  34                AutoClientDMAAP.aafPassword = aafPassword;
  35         }
  36
  37         public void setAuto(NotificationScheme scheme,
  38                         NotificationHandler handler) {
  39                 AutoClientDMAAP.scheme = scheme;
  40                 AutoClientDMAAP.handler = handler;
  41         }
  42
  43         public static void setScheme(NotificationScheme scheme) {
  44                 AutoClientDMAAP.scheme = scheme;
  45         }
  46
  47         public static boolean getStatus(){
  48                 return AutoClientDMAAP.status;
  49         }
  50
  51         public static String getTopic() {
  52                 return AutoClientDMAAP.topic;
  53         }
  54
  55         public static String getNotficationType(){
  56                 return AutoClientDMAAP.notficatioinType;
  57         }
  58
  59         public synchronized boolean isRunning() {
  60                 return this.isRunning;
  61         }
  62
  63         public synchronized void terminate() {
  64                 this.isRunning = false;
  65         }
  66
  67         @Override
  68         public void run() {
  69                 synchronized(this) {
  70                         this.isRunning = true;
  71                 }
  72                 String group =  UUID.randomUUID ().toString ();
  73                 String id = "0";
  74
  75                 // Stop and Start needs to be done.
  76                 if (scheme != null && handler!=null) {
  77                         if (scheme.equals(NotificationScheme.AUTO_ALL_NOTIFICATIONS) || scheme.equals(NotificationScheme.AUTO_NOTIFICATIONS)) {
  78
  79                                 // create a loop to listen for messages from DMaaP server
  80                                 try {
  81                                         dmaapConsumer = new BusConsumer.DmaapConsumerWrapper(dmaapList, topic, aafLogin, aafPassword, group, id, 15*1000, 1000 );
  82                                 } catch (Exception e) {
  83                                         logger.error(XACMLErrorConstants.ERROR_PROCESS_FLOW + "Unable to create DMaaP Consumer: ", e);
  84                                 }
  85
  86                                 while (this.isRunning() )
  87                                 {
  88                                         try {
  89                                                 for ( String msg : dmaapConsumer.fetch () )
  90                                                 {
  91                                                         logger.debug("Auto Notification Recieved Message " + msg + " from DMAAP server : " + dmaapList.toString());
  92                                                         notification = NotificationUnMarshal.notificationJSON(msg);
  93                                                         callHandler();
  94                                                 }
  95                                         } catch (Exception e) {
  96                                                 logger.debug("Error in processing DMAAP message");
  97                                         }
  98
  99                                 }
 100                                 logger.debug("Stopping DMAAP Consumer loop will no longer fetch messages from the servers");
 101                         }
 102                 }
 103         }
 104
 105         private static void callHandler() {
 106                 if (handler != null && scheme != null) {
 107                         if (scheme.equals(NotificationScheme.AUTO_ALL_NOTIFICATIONS)) {
 108                                 boolean removed = false, updated = false;
 109                                 if (notification.getRemovedPolicies() != null && !notification.getRemovedPolicies().isEmpty()) {
 110                                         removed = true;
 111                                 }
 112                                 if (notification.getLoadedPolicies() != null && !notification.getLoadedPolicies().isEmpty()) {
 113                                         updated = true;
 114                                 }
 115                                 if (removed && updated) {
 116                                         notification.setNotificationType(NotificationType.BOTH);
 117                                 } else if (removed) {
 118                                         notification.setNotificationType(NotificationType.REMOVE);
 119                                 } else if (updated) {
 120                                         notification.setNotificationType(NotificationType.UPDATE);
 121                                 }
 122                                 handler.notificationReceived(notification);
 123                         } else if (scheme.equals(NotificationScheme.AUTO_NOTIFICATIONS)) {
 124                                 PDPNotification newNotification = MatchStore.checkMatch(notification);
 125                                 if (newNotification.getNotificationType() != null) {
 126                                         handler.notificationReceived(newNotification);
 127                                 }
 128                         }
 129                 }
 130         }
 131
 132 }