added streaming api and test cases

Author: janithb

src/main/java/org/audit4j/core/handler/file/ArchiveManager.java

@@ -30,18 +30,18 @@ public void run() {
 	public Integer extractArchiveDateCount(String datePattern) {
 		int dateCount = 0;
 		String[] splits = datePattern.split("d|M|y");
-		if (splits.length>1) {
+		if (splits.length>0) {
 			dateCount = dateCount + Integer.valueOf(splits[0]);
 		}
 
 		System.out.println(dateCount);
 
-		if (splits.length>2) {
+		if (splits.length>1) {
 			dateCount = dateCount + (Integer.valueOf(splits[1]) * 30);
 		}
 		System.out.println(dateCount);
 
-		if (splits.length>3) {
+		if (splits.length>2) {
 			dateCount = dateCount + (Integer.valueOf(splits[2]) * 365);
 		}
 		System.out.println(dateCount);

src/main/java/org/audit4j/core/io/AsyncAuditOutputStream.java

@@ -0,0 +1,48 @@
+package org.audit4j.core.io;
+
+import org.audit4j.core.dto.AuditEvent;
+
+import reactor.core.Environment;
+import reactor.core.composable.Deferred;
+import reactor.core.composable.Stream;
+import reactor.core.composable.spec.Streams;
+import reactor.function.Consumer;
+import reactor.function.support.Boundary;
+
+public class AsyncAuditOutputStream implements AuditOutputStream {
+	final AuditOutputStream outputStream;
+	Deferred<AuditEvent, Stream<AuditEvent>> deferred = null;
+	static final Environment ENV = new Environment();
+	Boundary b = null;
+
+	public AsyncAuditOutputStream(final AuditOutputStream outputStream) {
+		this.outputStream = outputStream;
+		b = new Boundary();
+		deferred = Streams.<AuditEvent> defer().env(ENV).dispatcher(Environment.RING_BUFFER).get();
+		Stream<AuditEvent> stream = deferred.compose();
+		stream.consume(b.bind(new Consumer<AuditEvent>() {
+			@Override
+			public void accept(AuditEvent event) {
+				outputStream.write(event);
+			}
+		}));
+	}
+
+	@Override
+	public AsyncAuditOutputStream write(AuditEvent event) {
+		deferred.accept(event);
+		b.await();
+		return this;
+	}
+
+	@Override
+	public void close() {
+		ENV.shutdown();
+		outputStream.close();
+	}
+
+	@Override
+	public Object clone() {
+		return null;
+	}
+}

src/main/java/org/audit4j/core/io/AuditEventOutputStream.java

@@ -19,48 +19,70 @@
 
 package org.audit4j.core.io;
 
+import org.audit4j.core.TroubleshootManager;
+import org.audit4j.core.ValidationManager;
 import org.audit4j.core.dto.AuditEvent;
+import org.audit4j.core.exception.TroubleshootException;
+import org.audit4j.core.exception.ValidationException;
 
 /**
  * The Class AuditEventOutputStream.
- *
+ * 
  * @author <a href="mailto:[email protected]">Janith Bandara</a>
  */
 public class AuditEventOutputStream implements AuditOutputStream {
 
-    /** The output stream. */
-    AuditOutputStream outputStream;
+	/** The output stream. */
+	AuditOutputStream outputStream;
+
+	/**
+	 * Instantiates a new audit event output stream.
+	 * 
+	 * @param outputStream
+	 *            the output stream
+	 */
+	public AuditEventOutputStream(AuditOutputStream outputStream) {
+		this.outputStream = outputStream;
+	}
 
-    /**
-     * Instantiates a new audit event output stream.
-     *
-     * @param outputStream the output stream
-     */
-    public AuditEventOutputStream(AuditOutputStream outputStream) {
-        this.outputStream = outputStream;
-    }
+	/*
+	 * (non-Javadoc)
+	 * 
+	 * @see
+	 * org.audit4j.core.io.AuditOutputStream#write(org.audit4j.core.dto.AuditEvent
+	 * )
+	 */
+	@Override
+	public AuditEventOutputStream write(AuditEvent event) {
+		try {
+			ValidationManager.validateEvent(event);
+			outputStream.write(event);
+		} catch (ValidationException e) {
+			try{
+			TroubleshootManager.troubleshootEvent(event);
+			outputStream.write(event);
+			}catch(TroubleshootException e2){
+				
+			}
+		}
+		
+		return this;
+	}
 
-    /* (non-Javadoc)
-     * @see org.audit4j.core.io.AuditOutputStream#write(org.audit4j.core.dto.AuditEvent)
-     */
-    @Override
-    public AuditEventOutputStream write(AuditEvent event) {
-        outputStream.write(event);
-        return this;
-    }
+	/*
+	 * (non-Javadoc)
+	 * 
+	 * @see org.audit4j.core.io.AuditOutputStream#close()
+	 */
+	@Override
+	public void close() {
+		outputStream.close();
+		outputStream = null;
+	}
 
-    /* (non-Javadoc)
-     * @see org.audit4j.core.io.AuditOutputStream#close()
-     */
-    @Override
-    public void close() {
-        outputStream.close();
-        outputStream = null;
-    }
-    
-    @Override
-    public Object clone(){
-    	return null;
-    }
+	@Override
+	public Object clone() {
+		return null;
+	}
 
 }

src/main/java/org/audit4j/core/io/AuditProcessOutputStream.java

@@ -0,0 +1,32 @@
+package org.audit4j.core.io;
+
+import org.audit4j.core.AuditEventProcessor;
+import org.audit4j.core.AuditProcessor;
+import org.audit4j.core.Context;
+import org.audit4j.core.dto.AuditEvent;
+
+public class AuditProcessOutputStream implements AuditOutputStream {
+	public Context context;
+	
+	public AuditProcessOutputStream(Context context){
+		this.context = context;
+	}
+	
+	@Override
+	public AuditProcessOutputStream write(AuditEvent event) {
+		AuditProcessor<AuditEvent> processor = AuditEventProcessor.getInstance();
+		processor.setConf(context.getConfig());
+		processor.process(event);
+		return this;
+	}
+
+	@Override
+	public void close() {
+
+	}
+
+	@Override
+	public Object clone() {
+		return null;
+	}
+}

src/test/java/org/audit4j/core/Audit4jCoreTestSuit.java

@@ -1,16 +1,17 @@
 package org.audit4j.core;
 
+import org.audit4j.core.handler.file.ArchiveManagerTest;
 import org.junit.runner.RunWith;
 import org.junit.runners.Suite;
 import org.junit.runners.Suite.SuiteClasses;
 
 /**
  * The Class Audit4jCoreTestSuit.
- *
+ * 
  * @author <a href="mailto:[email protected]">Janith Bandara</a>
  */
 @RunWith(Suite.class)
-@SuiteClasses({ ValidationManagerTest.class, EncryptionUtilTest.class, TroubleshootManagerTest.class })
+@SuiteClasses({ ValidationManagerTest.class, EncryptionUtilTest.class, TroubleshootManagerTest.class, SmokeTest.class, ArchiveManagerTest.class })
 public class Audit4jCoreTestSuit {
 
 }

src/test/java/org/audit4j/core/LoadTest.java

@@ -0,0 +1,86 @@
+package org.audit4j.core;
+
+import java.util.concurrent.TimeUnit;
+
+import org.audit4j.core.dto.EventBuilder;
+import org.audit4j.core.io.AsyncAuditOutputStream;
+import org.audit4j.core.io.AuditEventOutputStream;
+import org.audit4j.core.io.AuditOutputStream;
+import org.audit4j.core.io.AuditProcessOutputStream;
+
+public class LoadTest {
+
+	public void testEvents_Param2_Load100(){
+		Long startTime = System.currentTimeMillis();
+		
+		for (int i = 0; i < 100; i++) {
+			String actor = "Dummy Actor" + i;
+			EventBuilder builder = new EventBuilder();
+			builder.addActor(actor).addAction("myMethod" + i).addOrigin("Origin" + i).addField("myParam1Name" + i, "param1" + i).addField("myParam2Name" + i, new Integer(1));
+			AuditManager manager = AuditManager.getInstance();
+			manager.audit(builder.build());
+		}
+		
+		Long endTime = System.currentTimeMillis();
+		Long totalTimeInMills =  endTime-startTime;
+		System.out.println("Total Time in mills: " + totalTimeInMills);
+	}
+	
+	public void testEvents_Param10_Load10000(){
+		Long startTime = System.currentTimeMillis();
+		
+		for (int i = 0; i < 10000; i++) {
+			String actor = "Dummy Actor" + i;
+			EventBuilder builder = new EventBuilder();
+			builder.addActor(actor).addAction("myMethod" + i).addOrigin("Origin" + i);
+			for (int j = 0; j < 10; j++) {
+				builder.addField("myParam1Name" + j + i, "param1" + j + i);
+			}
+			
+			AuditManager manager = AuditManager.getInstance();
+			manager.audit(builder.build());
+		}
+		
+		Long endTime = System.currentTimeMillis();
+		Long totalTimeInMills =  endTime-startTime;
+		System.out.println("Total Time in mills: " + totalTimeInMills);
+	}
+
+	
+	public void testStream_Param10_Load10000(){
+		Long startTime = System.currentTimeMillis();
+		Context context = new Context();
+		context.init();
+		
+		AsyncAuditOutputStream asyncStream = new AsyncAuditOutputStream(new AuditProcessOutputStream(context));
+		AuditOutputStream stream = new AuditEventOutputStream(asyncStream);
+
+		for (int i = 0; i < 10000; i++) {
+			String actor = "Dummy Actor" + i;
+			EventBuilder builder = new EventBuilder();
+			builder.addActor(actor).addAction("myMethod" + i).addOrigin("Origin" + i);
+			for (int j = 0; j < 10; j++) {
+				builder.addField("myParam1Name" + j + i, "param1" + j + i);
+			}
+			
+			stream.write(builder.build());
+			Long endTime = System.currentTimeMillis();
+			Long totalTimeInMills =  endTime-startTime;
+			System.out.println("Total Time in mills for this: " + totalTimeInMills);
+			
+		}
+		
+		
+		try {
+			TimeUnit.SECONDS.sleep(20);
+		} catch (InterruptedException e) {
+			// TODO Auto-generated catch block
+			e.printStackTrace();
+		}
+	}
+	
+	public static void main(String[] args) {
+		LoadTest test = new LoadTest();
+		test.testEvents_Param10_Load10000();
+	}
+}

src/test/java/org/audit4j/core/SmokeTest.java

@@ -0,0 +1,20 @@
+package org.audit4j.core;
+
+import java.util.concurrent.TimeUnit;
+
+import org.audit4j.core.dto.EventBuilder;
+import org.junit.Test;
+
+public class SmokeTest {
+
+	@Test
+	public void smokeTestAuditEvent() throws InterruptedException{
+		String actor = "Dummy Actor";
+		EventBuilder builder = new EventBuilder();
+		builder.addActor(actor).addAction("myMethod").addOrigin("Origin1").addField("myParam1Name", "param1").addField("myParam2Name", new Integer(2));
+		AuditManager manager = AuditManager.getInstance();
+		manager.audit(builder.build());
+		
+		TimeUnit.SECONDS.sleep(4);
+	}
+}