package com.box.sdk;

import java.util.ArrayList;
import java.util.Collection;
import java.util.LinkedHashSet;

import com.eclipsesource.json.JsonArray;
import com.eclipsesource.json.JsonObject;
import com.eclipsesource.json.JsonValue;

public class EventStream {
    private static final int LIMIT = 800;
    private static final int LRU_SIZE = 512;
    private static final URLTemplate EVENT_URL = new URLTemplate("events?limit=" + LIMIT + "&stream_position=%s");

    private final BoxAPIConnection api;
    private final Collection<EventListener> listeners;
    private final Object listenerLock;
    private final LinkedHashSet<String> receivedEvents;

    private boolean started;
    private Poller poller;
    private Thread pollerThread;

    public EventStream(BoxAPIConnection api) {
        this.api = api;
        this.listeners = new ArrayList<EventListener>();
        this.listenerLock = new Object();
        this.receivedEvents = new LinkedHashSet<String>(LRU_SIZE);
    }

    public void addListener(EventListener listener) {
        synchronized (this.listenerLock) {
            this.listeners.add(listener);
        }
    }

    public boolean isStarted() {
        return this.started;
    }

    public void stop() {
        if (!this.started) {
            throw new IllegalStateException("Cannot stop the EventStream because it isn't started.");
        }

        this.started = false;
        this.pollerThread.interrupt();
    }

    public void start() {
        if (this.started) {
            throw new IllegalStateException("Cannot start the EventStream because it isn't stopped.");
        }

        BoxAPIRequest request = new BoxAPIRequest(this.api, EVENT_URL.build(this.api.getBaseURL(), "now"), "GET");
        BoxJSONResponse response = (BoxJSONResponse) request.send();
        JsonObject jsonObject = JsonObject.readFrom(response.getJSON());
        final long initialPosition = jsonObject.get("next_stream_position").asLong();
        this.poller = new Poller(initialPosition);

        this.pollerThread = new Thread(this.poller);
        this.pollerThread.setUncaughtExceptionHandler(new Thread.UncaughtExceptionHandler() {
            public void uncaughtException(Thread t, Throwable e) {
                EventStream.this.notifyException(e);
            }
        });
        this.pollerThread.start();

        this.started = true;
    }

    private void notifyEvent(BoxEvent event) {
        synchronized (this.listenerLock) {
            boolean newEvent = this.receivedEvents.add(event.getID());
            if (newEvent) {
                if (this.receivedEvents.size() > LRU_SIZE) {
                    this.receivedEvents.iterator().remove();
                }

                for (EventListener listener : this.listeners) {
                    listener.onEvent(event);
                }
            }
        }
    }

    private void notifyException(Throwable e) {
        if (e instanceof InterruptedException && !this.started) {
            return;
        }

        this.stop();
        synchronized (this.listenerLock) {
            for (EventListener listener : this.listeners) {
                if (listener.onException(e)) {
                    return;
                }
            }
        }
    }

    private class Poller implements Runnable {
        private final long initialPosition;

        private RealtimeServerConnection server;

        public Poller(long initialPosition) {
            this.initialPosition = initialPosition;
            this.server = new RealtimeServerConnection(EventStream.this.api);
        }

        @Override
        public void run() {
            long position = this.initialPosition;
            while (!Thread.interrupted()) {
                if (this.server.getRemainingRetries() == 0) {
                    this.server = new RealtimeServerConnection(EventStream.this.api);
                }

                if (this.server.waitForChange(position)) {
                    if (Thread.interrupted()) {
                        return;
                    }

                    BoxAPIRequest request = new BoxAPIRequest(EventStream.this.api,
                        EVENT_URL.build(EventStream.this.api.getBaseURL(), position), "GET");
                    BoxJSONResponse response = (BoxJSONResponse) request.send();
                    JsonObject jsonObject = JsonObject.readFrom(response.getJSON());
                    position = jsonObject.get("next_stream_position").asLong();
                    JsonArray entriesArray = jsonObject.get("entries").asArray();
                    for (JsonValue entry : entriesArray) {
                        BoxEvent event = new BoxEvent(EventStream.this.api, entry.asObject());
                        EventStream.this.notifyEvent(event);
                    }
                }
            }
        }
    }
}

@@ -230,16 +230,551 @@
 Value;%0A%0A
+/**%0A * Receives real-time events from the API and forwards them to %7B@link EventListener EventListeners%7D.%0A *%0A * %3Cp%3EThis class handles long polling the Box events endpoint in order to receive real-time user or enterprise events.%0A * When an EventStream is started, it begins long polling on a separate thread until the %7B@link #stop%7D method is called.%0A * Since the API may return duplicate events, EventStream also maintains a small cache of the most recently received%0A * event IDs in order to automatically deduplicate events.%3C/p%3E%0A *%0A */%0A
 public c
@@ -1276,16 +1276,136 @@
 hread;%0A%0A
+    /**%0A     * Constructs an EventStream using an API connection.%0A     * @param  api the API connection to use.%0A     */%0A
     publ
@@ -1632,32 +1632,164 @@
 U_SIZE);%0A    %7D%0A%0A
+    /**%0A     * Adds a listener that will be notified when an event is received.%0A     * @param listener the listener to add.%0A     */%0A
     public void 
@@ -1936,72 +1936,382 @@
     
-public boolean isStarted() %7B%0A        return this.started;%0A    %7D%0A
+/**%0A     * Indicates whether or not this EventStream has been started.%0A     * @return true if this EventStream has been started; otherwise false.%0A     */%0A    public boolean isStarted() %7B%0A        return this.started;%0A    %7D%0A%0A    /**%0A     * Stops this EventStream and disconnects from the API.%0A     * @throws IllegalStateException if the EventStream is already stopped.%0A     */
 %0A   
@@ -2546,24 +2546,180 @@
 t();%0A    %7D%0A%0A
+    /**%0A     * Starts this EventStream and begins long polling the API.%0A     * @throws IllegalStateException if the EventStream is already started.%0A     */%0A
     public v

Add documentation for EventStream