package com.github.dockerjava.jaxrs;

import static com.google.common.base.Preconditions.checkNotNull;

import java.io.InputStream;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import javax.ws.rs.client.WebTarget;
import javax.ws.rs.core.Response;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.fasterxml.jackson.core.JsonFactory;
import com.fasterxml.jackson.core.JsonParser;
import com.fasterxml.jackson.core.JsonToken;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.github.dockerjava.api.command.EventCallback;
import com.github.dockerjava.api.command.EventsCmd;
import com.github.dockerjava.api.model.Event;
import com.github.dockerjava.jaxrs.util.WrappedResponseInputStream;

public class EventsCmdExec extends AbstrDockerCmdExec<EventsCmd, ExecutorService> implements EventsCmd.Exec {
    private static final Logger LOGGER = LoggerFactory.getLogger(EventsCmdExec.class);
    
    public EventsCmdExec(WebTarget baseResource) {
        super(baseResource);
    }

    @Override
    protected ExecutorService execute(EventsCmd command) {
    	ExecutorService executorService = Executors.newSingleThreadExecutor();
    	
        WebTarget webResource = getBaseResource().path("/events")
                .queryParam("since", command.getSince())
                .queryParam("until", command.getUntil());

        LOGGER.trace("GET: {}", webResource);
        EventNotifier eventNotifier = EventNotifier.create(command.getEventCallback(), webResource);
        executorService.submit(eventNotifier);
        return executorService;
    }
    
    private static class EventNotifier implements Callable<Void> {
        private static final JsonFactory JSON_FACTORY = new JsonFactory();
        private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();

        private final EventCallback eventCallback;
        private final WebTarget webTarget;

        private EventNotifier(EventCallback eventCallback, WebTarget webTarget) {
            this.eventCallback = eventCallback;
            this.webTarget = webTarget;
        }

        public static EventNotifier create(EventCallback eventCallback, WebTarget webTarget) {
            checkNotNull(eventCallback, "An EventCallback must be provided");
            checkNotNull(webTarget, "An WebTarget must be provided");
            return new EventNotifier(eventCallback, webTarget);
        }

        @Override
        public Void call() throws Exception {
            int numEvents=0;
            Response response = null;
            try {
                response = webTarget.request().get(Response.class);
                InputStream inputStream = new WrappedResponseInputStream(response);
                JsonParser jp = JSON_FACTORY.createParser(inputStream);
                while (jp.nextToken() != JsonToken.END_OBJECT && !jp.isClosed() && eventCallback.isReceiving()) {
                    eventCallback.onEvent(OBJECT_MAPPER.readValue(jp, Event.class));
                    numEvents++;
                }
            }
            catch(Exception e) {
                eventCallback.onException(e);
            }
            finally {
                if (response != null) {
                    response.close();
                }
            }
            eventCallback.onCompletion(numEvents);
            return null;
        }
    }
}

@@ -2968,32 +2968,59 @@
                 
+try %7B%0A                    %09
 eventCallback.on
@@ -3068,16 +3068,132 @@
 lass));%0A
+                    %7D catch(Exception e) %7B%0A                    %09eventCallback.onException(e);%0A                    %7D%0A
         

Fix issue #201