1 package org.openecomp.policy.std;
5 import org.json.JSONObject;
6 import org.openecomp.policy.api.NotificationScheme;
7 import org.openecomp.policy.api.NotificationType;
8 import org.openecomp.policy.api.PDPNotification;
9 import org.openecomp.policy.common.logging.flexlogger.FlexLogger;
10 import org.openecomp.policy.common.logging.flexlogger.Logger;
11 import org.openecomp.policy.utils.BusConsumer;
12 import org.openecomp.policy.utils.BusPublisher;
13 import org.openecomp.policy.xacml.api.XACMLErrorConstants;
15 public class ManualClientEndDMAAP {
16 private static StdPDPNotification notification = null;
17 private static String resultJson = null;
18 private static Logger logger = FlexLogger.getLogger(ManualClientEndDMAAP.class.getName());
19 private static BusConsumer dmaapConsumer = null;
20 private static String uniquID = null;
21 private static String topic = null;
24 public static PDPNotification result(NotificationScheme scheme) {
25 if (resultJson == null || notification == null) {
26 logger.debug("No Result" );
29 if(scheme.equals(NotificationScheme.MANUAL_ALL_NOTIFICATIONS)) {
30 boolean removed = false, updated = false;
31 if(notification.getRemovedPolicies()!=null && !notification.getRemovedPolicies().isEmpty()){
34 if(notification.getLoadedPolicies()!=null && !notification.getLoadedPolicies().isEmpty()){
37 if(removed && updated) {
38 notification.setNotificationType(NotificationType.BOTH);
40 notification.setNotificationType(NotificationType.REMOVE);
42 notification.setNotificationType(NotificationType.UPDATE);
45 }else if(scheme.equals(NotificationScheme.MANUAL_NOTIFICATIONS)) {
46 return MatchStore.checkMatch(notification);
53 private static void publishMessage(String pubTopic, String uniqueID, List<String> dmaapList, String aafLogin, String aafPassword) {
54 BusPublisher pub = null;
56 pub = new BusPublisher.DmaapPublisherWrapper(dmaapList, topic, aafLogin, aafPassword);
57 final JSONObject msg1 = new JSONObject ();
58 msg1.put ( "JSON", "DMaaP Update Request UID=" + uniqueID);
59 pub.send ( "MyPartitionKey", msg1.toString () );
60 } catch (Exception e) {
61 logger.error(XACMLErrorConstants.ERROR_PROCESS_FLOW + "Unable to create DMaaP Publisher: ", e);
68 //NOTE: should be able to remove this for DMAAP since we will not be creating topics dynamically
69 public static void createTopic (String topic, String uniquID, List<String> dmaapList, String aafLogin, String aafPassword){
70 ManualClientEndDMAAP.topic = topic;
71 publishMessage(topic, uniquID, dmaapList, aafLogin, aafPassword);
75 public static void start(List<String> dmaapList, String topic, String aafLogin, String aafPassword, String uniqueID) {
77 ManualClientEndDMAAP.uniquID = uniqueID;
78 ManualClientEndDMAAP.topic = topic;
83 dmaapConsumer = new BusConsumer.DmaapConsumerWrapper(dmaapList, topic, aafLogin, aafPassword, "clientGroup", id, 15*1000, 1000);
84 } catch (Exception e) {
85 logger.error(XACMLErrorConstants.ERROR_PROCESS_FLOW + "Unable to create DMaaP Consumer: ", e);
90 publishMessage(topic, uniquID, dmaapList, aafLogin, aafPassword);
92 for ( String msg : dmaapConsumer.fetch () )
94 logger.debug("Manual Notification Recieved Message " + msg + " from DMaaP server : " + dmaapList.toString());
96 if (!msg.contains("DMaaP Update")){
97 notification = NotificationUnMarshal.notificationJSON(msg);
101 }catch (Exception e) {
102 logger.error(XACMLErrorConstants.ERROR_PROCESS_FLOW + "Unable to fetch messages from DMaaP servers: ", e);