Added methods for event consuming.

Author: Joerg Huber

SIF3InfraREST/SIF3Demo/sif3-demo-web/src/main/java/systemic/sif3/demo/rest/consumer/functional/BaseFunctionalConsumer.java

@@ -17,10 +17,22 @@
 
 package systemic.sif3.demo.rest.consumer.functional;
 
+import java.io.BufferedWriter;
+import java.io.FileWriter;
+import java.util.Date;
+import java.util.HashMap;
+
+import javax.ws.rs.core.MediaType;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import au.com.systemic.framework.utils.DateUtils;
+import sif3.common.model.EventMetadata;
 import sif3.common.model.PagingInfo;
+import sif3.common.model.SIFContext;
+import sif3.common.model.SIFEvent;
+import sif3.common.model.SIFZone;
 import sif3.common.model.delayed.DelayedResponseReceipt;
 import sif3.common.model.job.PhaseInfo;
 import sif3.common.utils.JAXBUtils;
@@ -42,7 +54,11 @@
 public abstract class BaseFunctionalConsumer extends AbstractFunctionalServiceConsumer
 {
     protected final Logger logger = LoggerFactory.getLogger(getClass());
-
+    
+    private final static String RECORD_MARKER = "\n================================================================================================\n";
+    private HashMap<String, Integer> recordCounters = new HashMap<String, Integer>();
+    private static Boolean writePayload = null;
+    
     public BaseFunctionalConsumer()
     {
         super();
@@ -125,4 +141,81 @@ public void processDelayedError(ErrorDetails error, DelayedResponseReceipt recei
     {
         logger.debug("Received DELAYED ERROR Response:\n"+error+"\nDelayed Receipt Details:\n"+receipt);
     }
+    
+    /* (non-Javadoc)
+     * @see sif3.infra.rest.consumer.AbstractFunctionalServiceConsumer#processJobEvent(sif3.common.model.SIFEvent, sif3.common.model.SIFZone, sif3.common.model.SIFContext, sif3.common.model.EventMetadata, java.lang.String, java.lang.String)
+     */
+    @Override
+    public void processJobEvent(SIFEvent<JobCollectionType> sifEvent, SIFZone zone, SIFContext context, EventMetadata metadata, String msgReadID, String consumerID)
+    {
+        String consumerName = getPrettyName()+"(QueueID:"+msgReadID+"; ConsumerID: "+consumerID+")";
+        logger.debug(consumerName +" received an event from Zone = "+zone+", Context = "+context+" and Event Metadata = "+metadata);
+        dumpJobEvent(sifEvent, zone, context, msgReadID, consumerID);
+    }
+    
+    /*---------------------*/
+    /*-- Private Methods --*/
+    /*---------------------*/
+    
+    private void dumpJobEvent(SIFEvent<JobCollectionType> sifEvent, SIFZone zone, SIFContext context, String msgReadID, String consumerID)
+    {
+//      logger.debug(consumerID +" write data to file...");
+        String filename = getFileNameOnly(consumerID);
+        String fullFileName= getFullFileName(consumerID);
+        Integer recordNum = recordCounters.get(filename);
+        if (recordNum == null)
+        {
+            recordNum = 1;
+            recordCounters.put(filename, recordNum);
+        }
+        String timestamp = DateUtils.getISO8601withSecFraction(new Date());
+
+        BufferedWriter out = null;
+        try
+        {
+            FileWriter fstream = new FileWriter(fullFileName, true);
+            out = new BufferedWriter(fstream);
+            out.write(RECORD_MARKER);
+            out.write("Record " + recordNum + " - processed by Thread ID = "+Thread.currentThread().getId()+"\n"+sifEvent.getListSize()+" "+sifEvent.getEventAction().name()+" Events from Queue Reader "+ msgReadID+"\nReceived at "+timestamp+" from Zone = "+zone.getId()+" and Context = "+context.getId());
+            out.write(RECORD_MARKER);
+            
+            if (writePayload == null)
+            {
+                writePayload = getServiceProperties().getPropertyAsBool("test.consumer.write.payload", true);
+            }
+            
+            if (writePayload)
+            {
+                out.write(getMarshaller().marshal(sifEvent.getSIFObjectList(), MediaType.APPLICATION_XML_TYPE));
+            }
+            recordNum++;
+            recordCounters.put(filename, recordNum);
+        }
+        catch (Exception ex)
+        {
+            logger.error("Failed to write data to dump file with name:" + fullFileName, ex);
+        }
+        finally
+        {
+            if (out != null)
+            {
+                try
+                {
+                    out.close();
+                }
+                catch (Exception ex) {} // nothing we can do
+            }
+        }
+    }
+    
+    private String getFullFileName(String consumerID)
+    {
+      return getServiceProperties().getPropertyAsString("test.tempDir.output","") + "/" + getFileNameOnly(consumerID);
+      
+    }
+    
+    private String getFileNameOnly(String consumerID)
+    {
+      return consumerID+".log";
+    }
 }

SIF3InfraREST/SIF3Demo/sif3-demo-web/src/main/java/systemic/sif3/demo/rest/consumer/functional/RolloverStudentConsumer.java

@@ -47,7 +47,6 @@ public String getServiceURLNameSingular()
         return "RolloverStudent";
     }
 
-
     /* (non-Javadoc)
      * @see sif3.infra.rest.consumer.BaseConsumer#shutdown()
      */