package com.thinkbiganalytics.nifi.provenance.model;

import org.joda.time.DateTime;

import java.util.concurrent.atomic.AtomicLong;

/**
 * Created by sr186054 on 8/15/16.
 */
public class ProcessorStats {

    private String processorId;

    public ProcessorStats(String processorId) {
        this.processorId = processorId;
    }

    private AtomicLong eventsCount = new AtomicLong(0L);

    private Double avgProcessTime;

    private Long totalDurationTime = 0L;

    private Long lastEventId;

    private DateTime lastProcessDate;

    public String getProcessorId() {
        return processorId;
    }


    public void addEvent(ProvenanceEventRecordDTO event) {
        Long totalEvents = eventsCount.incrementAndGet();
        lastEventId = event.getEventId();
        //?? store the last n events?
        lastProcessDate = new DateTime(event.getEventTime());
        totalDurationTime += event.getEventDuration();
        avgProcessTime = totalDurationTime.doubleValue() / totalEvents.doubleValue();


    }

    public AtomicLong getEventsCount() {
        return eventsCount;
    }

    public Double getAvgProcessTime() {
        return avgProcessTime;
    }

    public Long getTotalDurationTime() {
        return totalDurationTime;
    }

    public Long getLastEventId() {
        return lastEventId;
    }

    public DateTime getLastProcessDate() {
        return lastProcessDate;
    }
}

@@ -238,102 +238,8 @@
 d;%0A%0A
-    public ProcessorStats(String processorId) %7B%0A        this.processorId = processorId;%0A    %7D%0A
 %0A   
@@ -297,44 +297,8 @@
 );%0A%0A
-    private Double avgProcessTime;%0A%0A
     
@@ -405,16 +405,747 @@
 sDate;%0A%0A
+%0A    public ProcessorStats(String processorId) %7B%0A        this.processorId = processorId;%0A    %7D%0A%0A%0A    public ProcessorStats(ProcessorStats p1, ProcessorStats p2) %7B%0A        Long total = p1.getEventsCount().get() + p2.getEventsCount().get();%0A        DateTime lastProcessDate = p1.getLastProcessDate().isAfter(p2.getLastProcessDate()) ? p1.getLastProcessDate() : p2.getLastProcessDate();%0A        Long lastEventId = p1.getLastEventId() %3E p2.getLastEventId() ? p1.getLastEventId() : p2.getLastEventId();%0A        this.eventsCount = new AtomicLong(total);%0A        this.lastEventId = lastEventId;%0A        this.lastProcessDate = lastProcessDate;%0A        this.totalDurationTime = p1.getTotalDurationTime() + p2.getTotalDurationTime();%0A    %7D%0A%0A
     publ
@@ -1279,27 +1279,8 @@
     
- Long totalEvents =
  eve
@@ -1353,46 +1353,8 @@
 ();%0A
-        //?? store the last n events?%0A
     
@@ -1470,96 +1470,8 @@
 ();%0A
-        avgProcessTime = totalDurationTime.doubleValue() / totalEvents.doubleValue();%0A%0A%0A
     
@@ -1608,22 +1608,59 @@
 urn 
-avgProcessTime
+totalDurationTime.doubleValue() / eventsCount.get()
 ;%0A  
@@ -1855,24 +1855,24 @@
 essDate() %7B%0A
-
         retu
@@ -1897,10 +1897,120 @@
 ;%0A    %7D%0A
+%0A%0A    public ProcessorStats join(ProcessorStats that) %7B%0A        return new ProcessorStats(this, that);%0A%0A    %7D%0A
 %7D%0A

update Processor stats to not store avg, but rather calculate it