package jacz.util.event.notification;

import jacz.util.concurrency.timer.Timer;
import jacz.util.concurrency.timer.TimerAction;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * This class handles the process of notifying events emitted by an emitter to the subscribed receivers
 */
class NotificationReceiverHandler implements TimerAction {

    private final NotificationReceiver notificationReceiver;

    private final String emitterID;

    /**
     * Delay for emitting notifications (null for no delay)
     */
    private final Long millis;

    private final double timeFactorAtEachEvent;

    private final int limit;

    /**
     * Amount of stored events
     */
    private int eventCount;

    /**
     * Lists of received messages (for non-grouped)
     */
    private List<List<Object>> nonGroupedMessages;

    /**
     * Set of nonGroupedMessages (for grouped).
     */
    private List<Object> groupedMessages;

    /**
     * Timer for emitting notifications (if we want to have a delay)
     */
    private final Timer timer;

    /**
     * Task executor for sequentially notifying events to the observers (plus, we decouple the notification from
     * the thread that causes the event)
     */
    private final ExecutorService sequentialExecutorService;


    NotificationReceiverHandler(NotificationReceiver notificationReceiver, String emitterID, Long millis, double timeFactorAtEachEvent, int limit, String threadName) {
        this.notificationReceiver = notificationReceiver;
        this.emitterID = emitterID;
        this.millis = (millis != null && millis < 1L) ? null : millis;
        if (this.millis != null) {
            timer = new Timer(millis, this, false, threadName + "/" + NotificationProcessor.class.getName());
        } else {
            timer = null;
        }
        this.timeFactorAtEachEvent = (timeFactorAtEachEvent > 1d) ? 1.0d : (timeFactorAtEachEvent < 0d ? 0d : timeFactorAtEachEvent);
        this.limit = (limit < 1) ? 1 : limit;
        eventCount = 0;
        nonGroupedMessages = new ArrayList<>();
        groupedMessages = new ArrayList<>();
        sequentialExecutorService = Executors.newSingleThreadExecutor();
    }

    synchronized void newEvent(Object... messages) {
        eventCount++;
        nonGroupedMessages.add(Arrays.asList(messages));
        groupedMessages.addAll(Arrays.asList(messages));
        if (eventCount == limit) {
            notifyReceiver();
            stopTimer();
        } else if (millis != null) {
            setTimer();
        }
    }

    private void setTimer() {
        if (timer.getState() != Timer.State.RUNNING) {
            timer.reset(millis);
        } else {
            timer.reset(timeFactorAtEachEvent);
        }
    }

    private void stopTimer() {
        if (timer != null) {
            timer.stop();
        }
    }

    private synchronized void notifyReceiver() {
        if (eventCount > 0) {
            final int eventCountCopy = eventCount;
            sequentialExecutorService.submit(new Runnable() {
                @Override
                public void run() {
                    notificationReceiver.newEvent(emitterID, eventCountCopy, new ArrayList<>(nonGroupedMessages), new ArrayList<>(groupedMessages));
                }
            });
//            sequentialTaskExecutor.executeTask(new Runnable() {
//                @Override
//                public void run() {
//                    notificationReceiver.newEvent(emitterID, eventCountCopy, new ArrayList<>(nonGroupedMessages), new ArrayList<>(groupedMessages));
//                }
//            });
            resetMessages();
            eventCount = 0;
        }
    }

    private void resetMessages() {
        groupedMessages.clear();
        nonGroupedMessages.clear();
    }

    @Override
    public synchronized Long wakeUp(Timer timer) {
        // check in case it has been stopped before
        if (timer.getState() == Timer.State.RUNNING) {
            notifyReceiver();
        }
        return 0L;
    }

    public synchronized void stop() {
        if (timer != null) {
            timer.kill();
        }
        sequentialExecutorService.shutdown();
    }
}

@@ -3109,217 +3109,115 @@
     
-sequentialExecutorService.submit(new Runnable() %7B%0A                @Override%0A                public void run() %7B%0A                    notificationReceiver.newEvent(emitterID, eventCountCopy, new ArrayList%3C%3E(nonG
+List%3CList%3CObject%3E%3E nonGroupedMessagesCopy = new ArrayList%3C%3E(nonGroupedMessages);%0A            List%3CObject%3E g
 roup
@@ -3218,34 +3218,38 @@
  groupedMessages
-),
+Copy =
  new ArrayList%3C%3E
@@ -3269,47 +3269,10 @@
 ges)
-)
 ;%0A
-                %7D%0A            %7D);%0A//
     
@@ -3293,20 +3293,16 @@
 tial
-Task
 Executor
 .exe
@@ -3301,125 +3301,28 @@
 utor
-.executeTask(new Runnable() %7B%0A//                @Override%0A//                public void run() %7B%0A//                   
+Service.submit(() -%3E
  not
@@ -3375,32 +3375,16 @@
 ntCopy, 
-new ArrayList%3C%3E(
 nonGroup
@@ -3397,81 +3397,34 @@
 ages
-), new ArrayList%3C%3E(groupedMessages));%0A//                %7D%0A//            %7D
+Copy, groupedMessagesCopy)
 );%0A 
@@ -3443,48 +3443,12 @@
 eset
-Messages
 ();%0A
-            eventCount = 0;%0A
     
@@ -3482,24 +3482,16 @@
 id reset
-Messages
 () %7B%0A   
@@ -3552,24 +3552,48 @@
 es.clear();%0A
+        eventCount = 0;%0A
     %7D%0A%0A    @

Fix bug in notification processor