add distribute function to EventWriter interface and new endpoint v2/event/distribute to resource

src/main/java/com/proofpoint/event/collector/EventCollectorStats.java

@@ -24,7 +24,7 @@
 public interface EventCollectorStats
 {
     // EventCollector.IncomingEvents.Count (Tags: eventType=blah, eventStatus=valid)
-    CounterStat incomingEvents(@Key("eventType") String eventType, @Key("eventStatus") EventStatus eventStatus);
+    CounterStat inboundEvents(@Key("eventType") String eventType, @Key("eventStatus") EventStatus eventStatus, @Key("processType") ProcessType processType);
 
     CounterStat outboundEvents(@Key("eventType") String eventType, @Key("flowId") String flowId, @Key("outboundStatus") Status status);
 
@@ -54,4 +54,16 @@ public String toString()
             return name().toLowerCase();
         }
     }
-}
+
+    public enum ProcessType
+    {
+        WRITE,
+        DISTRIBUTE;
+
+        @Override
+        public String toString()
+        {
+            return name().toLowerCase();
+        }
+    }
+}

src/main/java/com/proofpoint/event/collector/EventResource.java

@@ -19,6 +19,7 @@
 import com.google.common.collect.ImmutableSet;
 import com.google.common.collect.Sets;
 import com.google.inject.Inject;
+import com.proofpoint.event.collector.EventCollectorStats.ProcessType;
 
 import javax.ws.rs.Consumes;
 import javax.ws.rs.POST;
@@ -33,40 +34,61 @@
 import static com.google.common.base.Preconditions.checkNotNull;
 import static com.proofpoint.event.collector.EventCollectorStats.EventStatus.UNSUPPORTED;
 import static com.proofpoint.event.collector.EventCollectorStats.EventStatus.VALID;
+import static com.proofpoint.event.collector.EventCollectorStats.ProcessType.DISTRIBUTE;
+import static com.proofpoint.event.collector.EventCollectorStats.ProcessType.WRITE;
+import static com.proofpoint.event.collector.EventResource.EventProcessor.DISTRIBUTOR;
+import static com.proofpoint.event.collector.EventResource.EventProcessor.WRITER;
 
 @Path("/v2/event")
 public class EventResource
 {
     private final Set<EventWriter> writers;
     private final Set<String> acceptedEventTypes;
+
     private final EventCollectorStats eventCollectorStats;
 
     @Inject
     public EventResource(Set<EventWriter> writers, ServerConfig config, EventCollectorStats eventCollectorStats)
     {
-        this.eventCollectorStats = eventCollectorStats;
+        this.eventCollectorStats = checkNotNull(eventCollectorStats, "eventCollectorStats is null");
         this.writers = checkNotNull(writers, "writers are null");
         this.acceptedEventTypes = ImmutableSet.copyOf(checkNotNull(config, "config is null").getAcceptedEventTypes());
     }
 
     @POST
     @Consumes(MediaType.APPLICATION_JSON)
-    public Response post(List<Event> events)
+    public Response write(List<Event> events)
+            throws IOException
+    {
+        return processEvents(WRITER, events, WRITE);
+    }
+
+    @POST
+    @Path("/distribute")
+    @Consumes(MediaType.APPLICATION_JSON)
+    public Response distribute(List<Event> events)
+            throws IOException
+    {
+        return processEvents(DISTRIBUTOR, events, DISTRIBUTE);
+    }
+
+    private Response processEvents(EventProcessor processor, List<Event> events, ProcessType processType)
             throws IOException
     {
         Set<String> badEvents = Sets.newHashSet();
         for (Event event : events) {
             if (acceptedEventType(event.getType())) {
+
                 for (EventWriter writer : writers) {
-                    writer.write(event);
+                    processor.process(writer, event);
                 }
 
-                eventCollectorStats.incomingEvents(event.getType(), VALID).add(1);
+                eventCollectorStats.inboundEvents(event.getType(), VALID, processType).add(1);
             }
             else {
                 badEvents.add(event.getType());
 
-                eventCollectorStats.incomingEvents(event.getType(), UNSUPPORTED).add(1);
+                eventCollectorStats.inboundEvents(event.getType(), UNSUPPORTED, processType).add(1);
             }
         }
 
@@ -82,4 +104,29 @@ private boolean acceptedEventType(String type)
     {
         return acceptedEventTypes.isEmpty() || acceptedEventTypes.contains(type);
     }
+
+    public enum EventProcessor
+    {
+        WRITER
+                {
+                    @Override
+                    void process(EventWriter writer, Event event)
+                            throws IOException
+                    {
+                        writer.write(event);
+                    }
+                },
+        DISTRIBUTOR
+                {
+                    @Override
+                    void process(EventWriter writer, Event event)
+                            throws IOException
+                    {
+                        writer.distribute(event);
+                    }
+                };
+
+        abstract void process(EventWriter writer, Event event)
+                throws IOException;
+    }
 }

src/main/java/com/proofpoint/event/collector/EventTapWriter.java

@@ -29,6 +29,7 @@
 import javax.annotation.PostConstruct;
 import javax.annotation.PreDestroy;
 import javax.inject.Inject;
+import java.io.IOException;
 import java.net.URI;
 import java.util.HashMap;
 import java.util.List;
@@ -160,6 +161,13 @@ public void write(Event event)
         }
     }
 
+    @Override
+    public void distribute(Event event)
+            throws IOException
+    {
+        write(event);
+    }
+
     private Map<String, Map<String, FlowInfo>> constructFlowInfoFromDiscovery()
     {
         List<ServiceDescriptor> descriptors = selector.selectAllServices();

src/main/java/com/proofpoint/event/collector/EventWriter.java

@@ -21,4 +21,7 @@ public interface EventWriter
 {
     void write(Event event)
             throws IOException;
+
+    void distribute(Event event)
+            throws IOException;
 }

src/main/java/com/proofpoint/event/collector/SpoolingEventWriter.java

@@ -123,6 +123,14 @@ public void write(Event event)
         counters.recordReceived(event.getType(), 1);
     }
 
+    @Override
+    public void distribute(Event event)
+            throws IOException
+    {
+        // This particular collector is not the first collector to see this event. As such, this event
+        // has already been uploaded to S3 and should not be uploaded again.
+    }
+
     public Map<String, CounterState> getCounts()
     {
         return counters.getCounts();

src/test/java/com/proofpoint/event/collector/InMemoryEventWriter.java

@@ -23,17 +23,30 @@
 
 public class InMemoryEventWriter implements EventWriter
 {
-    List<Event> events = Lists.newArrayList();
+    List<Event> writtenEvents = Lists.newArrayList();
+    List<Event> distributedEvents = Lists.newArrayList();
 
     @Override
     public void write(Event event)
             throws IOException
     {
-        this.events.add(event);
+        this.writtenEvents.add(event);
     }
 
-    public List<Event> getEvents()
+    @Override
+    public void distribute(Event event)
+            throws IOException
+    {
+        this.distributedEvents.add(event);
+    }
+
+    public List<Event> getWrittenEvents()
+    {
+        return ImmutableList.copyOf(writtenEvents);
+    }
+
+    public List<Event> getDistributedEvents()
     {
-        return ImmutableList.copyOf(events);
+        return ImmutableList.copyOf(distributedEvents);
     }
 }