Events API: Switch to a callback approach rather than a BlockingQueue

docker-java · marcuslinke · Sep 22, 2014 · Sep 18, 2014 · Sep 18, 2014 · Sep 21, 2014
commit 3d2a3f017f976e8b740ccd653c75b9755706dc98
diff --git a/src/main/java/com/github/dockerjava/api/DockerClient.java b/src/main/java/com/github/dockerjava/api/DockerClient.java
@@ -10,6 +10,7 @@
 import com.github.dockerjava.api.command.CopyFileFromContainerCmd;
 import com.github.dockerjava.api.command.CreateContainerCmd;
 import com.github.dockerjava.api.command.CreateImageCmd;
+import com.github.dockerjava.api.command.EventCallback;
 import com.github.dockerjava.api.command.EventsCmd;
 import com.github.dockerjava.api.command.InfoCmd;
 import com.github.dockerjava.api.command.InspectContainerCmd;
@@ -115,7 +116,7 @@ public CopyFileFromContainerCmd copyFileFromContainerCmd(
 
 	public UnpauseContainerCmd unpauseContainerCmd(String containerId);
 
-	public EventsCmd eventsCmd();
+	public EventsCmd eventsCmd(EventCallback eventCallback);
 
 	public void close() throws IOException;
 

diff --git a/src/main/java/com/github/dockerjava/api/command/EventCallback.java b/src/main/java/com/github/dockerjava/api/command/EventCallback.java
@@ -0,0 +1,10 @@
+package com.github.dockerjava.api.command;
+
+import com.github.dockerjava.api.model.Event;
+
+/**
+ * Event callback
+ */
+public interface EventCallback {
+    public void onEvent(Event event);
+}
diff --git a/src/main/java/com/github/dockerjava/api/command/EventsCmd.java b/src/main/java/com/github/dockerjava/api/command/EventsCmd.java
@@ -1,14 +1,14 @@
 package com.github.dockerjava.api.command;
 
-import com.github.dockerjava.api.model.EventStream;
+import com.github.dockerjava.api.model.EventNotifier;
 
 /**
  * Get events
  *
  * @param since - Show all events created since timestamp
  * @param until - Stream events until this timestamp
  */
-public interface EventsCmd extends DockerCmd<EventStream> {
+public interface EventsCmd extends DockerCmd<EventNotifier> {
     public EventsCmd withSince(String since);
 
     public EventsCmd withUntil(String until);
@@ -17,6 +17,8 @@ public interface EventsCmd extends DockerCmd<EventStream> {
 
     public String getUntil();
 
-    public static interface Exec extends DockerCmdExec<EventsCmd, EventStream> {
+    public EventCallback getEventCallback();
+
+    public static interface Exec extends DockerCmdExec<EventsCmd, EventNotifier> {
     }
 }
diff --git a/src/main/java/com/github/dockerjava/api/model/EventNotifier.java b/src/main/java/com/github/dockerjava/api/model/EventNotifier.java
@@ -0,0 +1,51 @@
+package com.github.dockerjava.api.model;
+
+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.google.common.base.Preconditions;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.concurrent.Callable;
+
+/**
+ * EventStream API
+ * <p/>
+ * Spawns a thread to poll for events to fill a BlockingQueue
+ */
+public class EventNotifier implements Closeable, Callable<Void> {
+    private static final JsonFactory JSON_FACTORY = new JsonFactory();
+    private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
+
+    private final EventCallback eventCallback;
+    private final InputStream inputStream;
+
+    private EventNotifier(EventCallback eventCallback, InputStream inputStream) {
+        this.eventCallback = eventCallback;
+        this.inputStream = inputStream;
+    }
+
+    public static EventNotifier create(EventCallback eventCallback, InputStream inputStream) {
+        Preconditions.checkNotNull(eventCallback, "An EventCallback must be provided");
+        Preconditions.checkNotNull(inputStream, "An InputStream must be provided");
+        return new EventNotifier(eventCallback, inputStream);
+    }
+
+    @Override
+    public void close() throws IOException {
+        inputStream.close();
+    }
+
+    @Override
+    public Void call() throws Exception {
+        JsonParser jp = JSON_FACTORY.createParser(inputStream);
+        while (jp.nextToken() != JsonToken.END_OBJECT && !jp.isClosed()) {
+            eventCallback.onEvent(OBJECT_MAPPER.readValue(jp, Event.class));
+        }
+        return null;
+    }
+}
diff --git a/src/main/java/com/github/dockerjava/api/model/EventStream.java b/src/main/java/com/github/dockerjava/api/model/EventStream.java
diff --git a/src/main/java/com/github/dockerjava/core/DockerClientImpl.java b/src/main/java/com/github/dockerjava/core/DockerClientImpl.java
@@ -253,8 +253,8 @@ public UnpauseContainerCmd unpauseContainerCmd(String containerId) {
 	}
 
     @Override
-    public EventsCmd eventsCmd() {
-        return new EventsCmdImpl(getDockerCmdExecFactory().createEventsCmdExec());
+    public EventsCmd eventsCmd(EventCallback eventCallback) {
+        return new EventsCmdImpl(getDockerCmdExecFactory().createEventsCmdExec(), eventCallback);
     }
 
     @Override

diff --git a/src/main/java/com/github/dockerjava/core/command/EventsCmdImpl.java b/src/main/java/com/github/dockerjava/core/command/EventsCmdImpl.java
@@ -1,20 +1,22 @@
 package com.github.dockerjava.core.command;
 
+import com.github.dockerjava.api.command.EventCallback;
 import com.github.dockerjava.api.command.EventsCmd;
-import com.github.dockerjava.api.model.EventStream;
-
-import java.io.InputStream;
+import com.github.dockerjava.api.model.EventNotifier;
 
 /**
  * Stream docker events
  */
-public class EventsCmdImpl extends AbstrDockerCmd<EventsCmd, EventStream> implements EventsCmd {
+public class EventsCmdImpl extends AbstrDockerCmd<EventsCmd, EventNotifier> implements EventsCmd {
+
+    private final EventCallback eventCallback;
 
     private String since;
     private String until;
 
-    public EventsCmdImpl(EventsCmd.Exec exec) {
+    public EventsCmdImpl(EventsCmd.Exec exec, EventCallback eventCallback) {
         super(exec);
+        this.eventCallback = eventCallback;
     }
 
     @Override
@@ -29,16 +31,23 @@ public EventsCmd withUntil(String until) {
         return this;
     }
 
+    @Override
     public String getSince() {
         return since;
     }
 
+    @Override
     public String getUntil() {
         return until;
     }
 
     @Override
-    public EventStream exec() {
+    public EventCallback getEventCallback() {
+        return eventCallback;
+    }
+
+    @Override
+    public EventNotifier exec() {
         return super.exec();
     }
 

diff --git a/src/main/java/com/github/dockerjava/jaxrs/EventsCmdExec.java b/src/main/java/com/github/dockerjava/jaxrs/EventsCmdExec.java
@@ -1,29 +1,29 @@
 package com.github.dockerjava.jaxrs;
 
 import com.github.dockerjava.api.command.EventsCmd;
-import com.github.dockerjava.api.model.EventStream;
+import com.github.dockerjava.api.model.EventNotifier;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import javax.ws.rs.client.WebTarget;
 import javax.ws.rs.core.Response;
 import java.io.InputStream;
 
-public class EventsCmdExec extends AbstrDockerCmdExec<EventsCmd, EventStream> implements EventsCmd.Exec {
+public class EventsCmdExec extends AbstrDockerCmdExec<EventsCmd, EventNotifier> implements EventsCmd.Exec {
     private static final Logger LOGGER = LoggerFactory.getLogger(EventsCmdExec.class);
 
     public EventsCmdExec(WebTarget baseResource) {
         super(baseResource);
     }
 
     @Override
-    public EventStream exec(EventsCmd command) {
+    protected EventNotifier execute(EventsCmd command) {
         WebTarget webResource = getBaseResource().path("/events")
                 .queryParam("since", command.getSince())
                 .queryParam("until", command.getUntil());
 
         LOGGER.trace("GET: {}", webResource);
         InputStream inputStream = webResource.request().get(Response.class).readEntity(InputStream.class);
-        return EventStream.create(inputStream);
+        return EventNotifier.create(command.getEventCallback(), inputStream);
     }
 }