Events can be published in parallel (#302)

Events can be published in parallel

Author: z-sztrom

CHANGELOG.md

@@ -1,6 +1,9 @@
+## 2.1.6
+- Events can be published in parallel.
+
 ## 2.1.2
 - Added external parameters to send username, password and/or uri to connect to Messagebus.
- 
+
 ## 2.1.1
 - Implemented the changes to log the eventId and HTTPStatus while the level is INFO.
 - Implemented the changes to print the user information while the log level is INFO.

pom.xml

@@ -9,7 +9,7 @@
         <version>2.0.12</version>
     </parent>
     <properties>
-        <eiffel-remrem-publish.version>2.1.5</eiffel-remrem-publish.version>
+        <eiffel-remrem-publish.version>2.1.6</eiffel-remrem-publish.version>
         <eiffel-remrem-semantics.version>2.4.0</eiffel-remrem-semantics.version>
     </properties>
     <artifactId>eiffel-remrem-publish</artifactId>

publish-common/src/main/java/com/ericsson/eiffel/remrem/publish/exception/RemRemPublishException.java

@@ -32,4 +32,15 @@ public RemRemPublishException(String message, RMQBeanConnectionFactory factory,
             Throwable cause) {
         super(message + factory.getHost() + ":" + factory.getPort(), cause);
     }
+
+    @Override
+    public String getMessage() {
+        String message = super.getMessage();
+        Throwable cause = getCause();
+        if (cause != null) {
+            message += "; root cause: '" + cause.getMessage() + "'";
+        }
+
+        return message;
+    }
 }

publish-common/src/main/java/com/ericsson/eiffel/remrem/publish/helper/RabbitMqProperties.java

@@ -318,7 +318,7 @@ public void init() {
             //The exception can be safely handled here as there is a check for existence of exchange is done before each publish.
             checkAndCreateExchangeIfNeeded();
         } catch (RemRemPublishException e) {
-            log.error("Error occured while setting up the RabbitMq Connection. "+e.getMessage());
+            log.error("Error occurred while setting up the RabbitMq Connection. "+e.getMessage());
             e.printStackTrace();
         }
 

publish-common/src/main/java/com/ericsson/eiffel/remrem/publish/service/EventTemplateHandler.java

@@ -71,6 +71,7 @@ public JsonNode eventTemplateParser(String jsonData , String eventName){
             updatedJson = mapper.readValue(eventTemplate, JsonNode.class);
         } catch (IOException e) {
             log.error(e.getMessage(), e);
+            return null;
         }
 
         // For each key/value pair for parsing to template

publish-common/src/main/java/com/ericsson/eiffel/remrem/publish/service/MessageService.java

@@ -60,10 +60,4 @@ public interface MessageService {
      * Does the cleanup like closing open connections
      */
     public void cleanUp(); 
-    
-    /**
-     * Implemented Status code for the response
-     */
-    public HttpStatus getHttpStatus();
-
 }

publish-common/src/main/java/com/ericsson/eiffel/remrem/publish/service/MessageServiceRMQImpl.java

@@ -15,6 +15,7 @@
 package com.ericsson.eiffel.remrem.publish.service;
 
 import java.io.IOException;
+import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -49,11 +50,6 @@
 
     Logger log = (Logger) LoggerFactory.getLogger(MessageServiceRMQImpl.class);
 
-    /*Variables handles status codes*/
-    List<Integer> statusCodes;
-    List<JsonElement> errorItems;
-    List<PublishResultItem> resultList;
-    boolean checkEventStatus;
     /*
      * (non-Javadoc)
      * @see com.ericsson.eiffel.remrem.publish.service.MessageService#send(java.util.Map, java.util.Map)
@@ -74,7 +70,6 @@ public SendResult send(Map<String, String> routingKeyMap, Map<String, String> ms
                 } else {
                     event = new PublishResultItem(entryKey,  HttpStatus.INTERNAL_SERVER_ERROR.value(), PropertiesConfig.SERVER_DOWN,
                             PropertiesConfig.SERVER_DOWN_MESSAGE);
-                    checkEventStatus = false;
                 }
                 } catch (NackException e) {
                     event = new PublishResultItem(entryKey, HttpStatus.INTERNAL_SERVER_ERROR.value(), PropertiesConfig.SERVER_DOWN,
@@ -105,65 +100,75 @@ public SendResult send(Map<String, String> routingKeyMap, Map<String, String> ms
      */
     @Override
     public SendResult send(String jsonContent, MsgService msgService, String userDomainSuffix, String tag, String routingKey) {
-
-        JsonParser parser = new JsonParser();
         try {
-            JsonElement json = parser.parse(jsonContent);
+            JsonElement json = JsonParser.parseString(jsonContent);
             if (json.isJsonArray()) {
                 return send(json, msgService, userDomainSuffix, tag, routingKey);
-            } else {
-                Map<String, String> map = new HashMap<>();
-                Map<String, String> routingKeyMap = new HashMap<>();
-                String eventId = msgService.getEventId(json.getAsJsonObject());
-                if (StringUtils.isNotBlank(eventId)) {
-                    String routing_key = PublishUtils.getRoutingKey(msgService, json.getAsJsonObject(), rmqHelper, userDomainSuffix, tag, routingKey);
-                    if (StringUtils.isNotBlank(routing_key)) {
-                        map.put(eventId, json.toString());
-                        routingKeyMap.put(eventId, routing_key);
-                    } else if (routing_key == null) {
-                        List<PublishResultItem> resultItemList = new CopyOnWriteArrayList<>();
-                        routingKeyGenerationFailure(resultItemList);
-                        return new SendResult(resultItemList);
-                    } else {
-                        List<PublishResultItem> resultItemList = new CopyOnWriteArrayList<>();
-                        PublishResultItem resultItem = rabbitmqConfigurationNotFound(msgService);
-                        resultItemList.add(resultItem);
-                        return new SendResult(resultItemList);
-                    }
+            }
+
+            Map<String, String> map = new HashMap<>();
+            Map<String, String> routingKeyMap = new HashMap<>();
+            String eventId = msgService.getEventId(json.getAsJsonObject());
+            if (StringUtils.isNotBlank(eventId)) {
+                String routing_key = PublishUtils.getRoutingKey(msgService, json.getAsJsonObject(), rmqHelper, userDomainSuffix, tag, routingKey);
+                if (StringUtils.isNotBlank(routing_key)) {
+                    map.put(eventId, json.toString());
+                    routingKeyMap.put(eventId, routing_key);
+                } else if (routing_key == null) {
+                    List<PublishResultItem> resultItemList = new ArrayList<>();
+                    routingKeyGenerationFailure(resultItemList);
+                    return new SendResult(resultItemList);
                 } else {
-                    List<PublishResultItem> resultItemList = new CopyOnWriteArrayList<>();
-                    createFailureResult(resultItemList);
+                    List<PublishResultItem> resultItemList = new ArrayList<>();
+                    PublishResultItem resultItem = rabbitmqConfigurationNotFound(msgService);
+                    resultItemList.add(resultItem);
                     return new SendResult(resultItemList);
                 }
-                return send(routingKeyMap, map, msgService);
+            } else {
+                List<PublishResultItem> resultItemList = new ArrayList<>();
+                createFailureResult(resultItemList);
+                return new SendResult(resultItemList);
             }
+            return send(routingKeyMap, map, msgService);
         } catch (final JsonSyntaxException e) {
             String resultMsg = "Could not parse JSON.";
             if (e.getCause() != null) {
                 resultMsg = resultMsg + " Cause: " + e.getCause().getMessage();
             }
             log.error(resultMsg, e.getMessage());
-            List<PublishResultItem> resultItemList = new CopyOnWriteArrayList<>();
+            List<PublishResultItem> resultItemList = new ArrayList<>();
             createFailureResult(resultItemList);
             return new SendResult(resultItemList);
         }
     }
 
+    protected SendResult createSendResult(List<PublishResultItem> resultList) {
+        SendResult result = new SendResult();
+        result.setEvents(resultList);
+        return result;
+    }
+
     /*
      * (non-Javadoc)
      * @see com.ericsson.eiffel.remrem.publish.service.MessageService#send(com.google.gson.JsonElement, com.ericsson.eiffel.remrem.protocol.MsgService, java.lang.String)
     */
     @Override
     public SendResult send(JsonElement json, MsgService msgService, String userDomainSuffix, String tag, String routingKey) {
+
+        List<PublishResultItem> resultList;
+        boolean checkEventStatus;
+
         Map<String, String> map = new HashMap<>();
         Map<String, String> routingKeyMap = new HashMap<>();
+
         SendResult result;
-        resultList = new CopyOnWriteArrayList<PublishResultItem>();
+        resultList = new ArrayList<>();
         if (json == null) {
             createFailureResult(resultList);
+            return createSendResult(resultList);
         }
+
         if (json.isJsonArray()) {
-            statusCodes = new CopyOnWriteArrayList<Integer>();
             checkEventStatus = true;
             JsonArray bodyJson = json.getAsJsonArray();
             for (JsonElement obj : bodyJson) {
@@ -174,57 +179,37 @@ public SendResult send(JsonElement json, MsgService msgService, String userDomai
                     if (StringUtils.isNotBlank(routing_key)) {
                         result = send(obj.toString(), msgService, userDomainSuffix, tag, routing_key);
                         resultList.addAll(result.getEvents());
-                        int statusCode = result.getEvents().get(0).getStatusCode();
-                        if (!statusCodes.contains(statusCode))
-                            statusCodes.add(statusCode);
                     } else if (routing_key == null) {
                         routingKeyGenerationFailure(resultList);
-                        errorItems = new CopyOnWriteArrayList<JsonElement>();
-                        int statusCode = resultList.get(0).getStatusCode();
-                        statusCodes.add(statusCode);
-                        errorItems.add(obj);
                         checkEventStatus = false;
                     } else {
                         PublishResultItem resultItem = rabbitmqConfigurationNotFound(msgService);
                         resultList.add(resultItem);
-                        int statusCode = resultItem.getStatusCode();
-                        statusCodes.add(statusCode);
                         break;
                     }
                 } else {
                     if (!checkEventStatus) {
-                        addUnsuccessfulResultItem(obj);
-                        int statusCode = resultList.get(0).getStatusCode();
-                        statusCodes.add(statusCode);
+                        addUnsuccessfulResultItem(resultList, obj);
                     } else {
                         createFailureResult(resultList);
-                        errorItems = new CopyOnWriteArrayList<JsonElement>();
-                        int statusCode = resultList.get(0).getStatusCode();
-                        statusCodes.add(statusCode);
-                        errorItems.add(obj);
                         checkEventStatus = false;
                     }
                 }
             }
         } else {
-            statusCodes = new CopyOnWriteArrayList<Integer>();
             result = send(json.toString(), msgService, userDomainSuffix, tag, routingKey);
             resultList.addAll(result.getEvents());
-            int statusCode = result.getEvents().get(0).getStatusCode();
-            if (!statusCodes.contains(statusCode))
-                statusCodes.add(statusCode);
         }
-        result = new SendResult();
-        result.setEvents(resultList);
-        return result;
+
+        return createSendResult(resultList);
     }
 
-    private String sendMessage(String routingKey, String msg, MsgService msgService) throws IOException,NackException, TimeoutException, RemRemPublishException {
+    private String sendMessage(String routingKey, String msg, MsgService msgService) throws IOException,TimeoutException, RemRemPublishException {
         String resultMsg = PropertiesConfig.SUCCESS;
         try {
             instantiateRmqHelper();
         } catch (RemRemPublishException e) {
-            log.error("RemRemPublishException occurred::" + e.getMessage());
+            log.error("RemRemPublishException occurred: {}", e.getMessage());
         }
         rmqHelper.send(routingKey, msg, msgService);
         return resultMsg;
@@ -268,7 +253,7 @@ private String getAndCheckEvent(MsgService msgService, Map<String, String> map,
 
     /**
      * Method returns result for the failure event.
-     * @param events for list the eiffel events results
+     * @param resultItemList for list the eiffel events results
      */
     private void createFailureResult(List<PublishResultItem> resultItemList) {
         PublishResultItem resultItem = new PublishResultItem(null, 400, PropertiesConfig.INVALID_MESSAGE,
@@ -293,21 +278,9 @@ private void routingKeyGenerationFailure(List<PublishResultItem> resultItemList)
         resultItemList.add(resultItem);
     }
 
-    private void addUnsuccessfulResultItem(JsonElement obj) {
+    private void addUnsuccessfulResultItem(List<PublishResultItem> resultList, JsonElement obj) {
         PublishResultItem event = new PublishResultItem(null, 503, PropertiesConfig.SERVICE_UNAVAILABLE,
                 PropertiesConfig.UNSUCCESSFUL_EVENT_CONTENT);
         resultList.add(event);
     }
-    
-    /**
-     * Method returns the Http response code for the events.
-     */
-    public HttpStatus getHttpStatus() {
-        if (statusCodes.size() > 1) {
-            return HttpStatus.MULTI_STATUS;
-        } else {
-            return HttpStatus.valueOf(statusCodes.get(0));
-            
-        }
-    }
 }