package com.thinkbiganalytics.nifi.provenance;

import com.thinkbiganalytics.nifi.provenance.collector.ProvenanceEventCollector;
import com.thinkbiganalytics.nifi.provenance.collector.ProvenanceEventFeedCollector;
import com.thinkbiganalytics.nifi.provenance.model.DelayedProvenanceEvent;
import com.thinkbiganalytics.nifi.provenance.model.ProvenanceEventRecordDTO;
import com.thinkbiganalytics.nifi.provenance.processor.ProvenanceEventProcessor;
import com.thinkbiganalytics.nifi.provenance.v2.cache.flowfile.ActiveFlowFile;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.stream.Collectors;

/**
 * Created by sr186054 on 8/14/16.
 */
public class DelayedProvenanceEventFeedConsumer implements Runnable {

    private StreamConfiguration configuration;

    private BlockingQueue<DelayedProvenanceEvent> queue;

    private ProvenanceEventCollector eventCollector;

    public DelayedProvenanceEventFeedConsumer(StreamConfiguration configuration, BlockingQueue<DelayedProvenanceEvent> queue) {
        this.queue = queue;
        this.configuration = configuration;
        //Group events by Feed before processing
        eventCollector = new ProvenanceEventFeedCollector();
    }


    private List<ProvenanceEventRecordDTO> takeAll() {
        List<DelayedProvenanceEvent> expired = new ArrayList<>();
        queue.drainTo(expired);
        return expired.stream().map(delayed -> delayed.getEvent()).collect(Collectors.toList());
    }

    public String streamingMapKey(ProvenanceEventRecordDTO event) {
        ActiveFlowFile flowFile = event.getFlowFile();
        if (flowFile != null && flowFile.getFirstEvent() != null) {
            return flowFile.getFirstEvent().getComponentId() + "-" + event.getComponentId();
        }
        return event.getComponentId();
    }

    private void process() {

        //everything in queue delayed, collect and process
        List<ProvenanceEventRecordDTO> events = takeAll();
        //collect the events into the correct grouping (by Feed)
        if (events != null && !events.isEmpty()) {
            //process each group of events
            //Grouped by root processorId and then by
            eventCollector.collect(events).entrySet().stream().forEach(entry -> {
                //This could be broken out into multiple threads if desired
                new ProvenanceEventProcessor(configuration).process(entry.getValue());
            });
        }


    }


    public void run() {
        while (true) {
            try {
                //Wait a period of time before processing to group all expired events together
                Thread.sleep(configuration.getProcessDelay());
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            process();
        }
    }


}

@@ -443,87 +443,8 @@
 sor;
-%0Aimport com.thinkbiganalytics.nifi.provenance.v2.cache.flowfile.ActiveFlowFile;
 %0A%0Aim
@@ -1425,347 +1425,8 @@
  %7D%0A%0A
-    public String streamingMapKey(ProvenanceEventRecordDTO event) %7B%0A        ActiveFlowFile flowFile = event.getFlowFile();%0A        if (flowFile != null && flowFile.getFirstEvent() != null) %7B%0A            return flowFile.getFirstEvent().getComponentId() + %22-%22 + event.getComponentId();%0A        %7D%0A        return event.getComponentId();%0A    %7D%0A
 %0A   
@@ -1485,17 +1485,8 @@
 ueue
- delayed,
  col
@@ -1766,19 +1766,50 @@
 rId 
-and then by
+with list of events from various FlowFiles
 %0A   
@@ -1963,16 +1963,154 @@
 desired%0A
+                //Send the List of events, grouped by Feed (root processorId) on to be processed to determine if they are stream or batch%0A
         

update comments