Merge branch 'development' of https://github.com/samagra-comms/outbound…

… into prod-development
samagra-comms · May 18, 2023 · 9787884 · 9787884
2 parents a05bcba + 9a439ed
commit 9787884
Showing 1 changed file with 20 additions and 2 deletions.
diff --git a/src/main/java/com/uci/outbound/consumers/OutboundKafkaController.java b/src/main/java/com/uci/outbound/consumers/OutboundKafkaController.java
@@ -52,17 +52,22 @@ public class OutboundKafkaController {
     @Value("${spring.mail.recipient}")
     private String recipient;
 
+    private static long count = 0;
+
     @EventListener(ApplicationStartedEvent.class)
     public void onMessage() {
+
         reactiveKafkaReceiver
                 .doOnNext(new Consumer<ReceiverRecord<String, String>>() {
                     @Override
                     public void accept(ReceiverRecord<String, String> msg) {
                         log.info("kafka message receieved!");
+                        final long startTime = System.nanoTime();
+                        logTimeTaken(startTime, 0, "process-start: %d ms");
                         XMessage currentXmsg = null;
                         try {
                             currentXmsg = XMessageParser.parse(new ByteArrayInputStream(msg.value().getBytes()));
-                            sendOutboundMessage(currentXmsg);
+                            sendOutboundMessage(currentXmsg, startTime);
                         } catch (Exception e) {
                             HashMap<String, String> attachments = new HashMap<>();
                             attachments.put("Exception", ExceptionUtils.getStackTrace(e));
@@ -90,7 +95,7 @@ public void accept(Throwable e) {
      * @param currentXmsg
      * @throws Exception
      */
-    public void sendOutboundMessage(XMessage currentXmsg) throws Exception {
+    public void sendOutboundMessage(XMessage currentXmsg, long startTime) throws Exception {
         String channel = currentXmsg.getChannelURI();
         String provider = currentXmsg.getProviderURI();
         IProvider iprovider = factoryProvider.getProvider(provider, channel);
@@ -125,6 +130,9 @@ public void accept(Throwable e) {
                                             @Override
                                             public void accept(XMessageDAO xMessageDAO) {
                                                 log.info("XMessage Object saved is with sent user ID >> " + xMessageDAO.getUserId());
+                                                count++;
+//                                                log.info("Insert Record in Cass : "+count);
+                                                logTimeTaken(startTime, 0, "Insert Record in Cass : " + count +" ::: process-end: %d ms");
                                             }
                                         });
                             } catch (Exception e) {
@@ -166,4 +174,14 @@ private void sentEmail(XMessage xMessage, String subject, String body, String re
 //        log.info("EmailDetails :" + emailDetails);
         emailService.sendMailWithAttachment(emailDetails);
     }
+
+    private void logTimeTaken(long startTime, int checkpointID, String formatedMsg) {
+        long endTime = System.nanoTime();
+        long duration = (endTime - startTime) / 1000000;
+        if(formatedMsg == null) {
+            log.info(String.format("CP-%d: %d ms", checkpointID, duration));
+        } else {
+            log.info(String.format(formatedMsg, duration));
+        }
+    }
 }