Merge branch '__rultor'

Author: rultor

src/main/java/com/amihaiemil/docker/Docker.java

@@ -26,12 +26,8 @@
 package com.amihaiemil.docker;
 
 import java.io.IOException;
-import java.util.stream.Stream;
-
 import org.apache.http.client.HttpClient;
 
-import javax.json.JsonObject;
-
 /**
  * Docker API entry point.
  * @author Mihai Andronache (amihaiemil@gmail.com)
@@ -48,25 +44,10 @@ public interface Docker {
     boolean ping() throws IOException;
 
     /**
-     * Read events from the server in real time. Pay attention:<br><br>
-     * The Stream is <b>infinite</b>, which means you have to specify a
-     * <b>limit</b> before calling a terminal operation on it. Otherwise,
-     * your terminal operation will run until the Server closes the connection.
-     * Example:
-     * <pre>
-     *   final Docker docker = ...;
-     *   final List&lt;JsonObject&gt; firstTen = docker.events()
-     *       .limit(10).collect(Collectors.toList());
-     *   //It will wait for and return the first 10 real-time events
-     *   //from the server.
-     * </pre>
-     * @return The events as a {@link Stream}.
-     * @throws IOException If an I/O error occurs.
-     * @throws UnexpectedResponseException If the API responds with an
-     *  unexpected status.
+     * Entry point for the Events API.
+     * @return Events.
      */
-    Stream<JsonObject> events()
-        throws IOException, UnexpectedResponseException;
+    Events events();
 
     /**
      * Entry point for the Containers API.

src/main/java/com/amihaiemil/docker/Events.java

@@ -0,0 +1,92 @@
+/**
+ * Copyright (c) 2018-2020, Mihai Emil Andronache
+ * All rights reserved.
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are met:
+ * 1)Redistributions of source code must retain the above copyright notice,
+ * this list of conditions and the following disclaimer.
+ * 2)Redistributions in binary form must reproduce the above copyright notice,
+ * this list of conditions and the following disclaimer in the documentation
+ * and/or other materials provided with the distribution.
+ * 3)Neither the name of docker-java-api nor the names of its
+ * contributors may be used to endorse or promote products derived from
+ * this software without specific prior written permission.
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
+ * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+ * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
+ * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
+ * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
+ * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
+ * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
+ * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
+ * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
+ * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
+ * POSSIBILITY OF SUCH DAMAGE.
+ */
+package com.amihaiemil.docker;
+
+import javax.json.JsonObject;
+import java.io.IOException;
+import java.time.LocalDateTime;
+import java.util.Map;
+import java.util.function.Supplier;
+import java.util.stream.Stream;
+
+/**
+ * Events API.
+ * @author Mihai Andronache (amihaiemil@gmail.com)
+ * @version $Id$
+ * @since 0.0.13
+ */
+public interface Events {
+
+    /**
+     * Show events created since the specified timestamp,
+     * then stream new events.
+     * @param timestamp Given timestamp.
+     * @return Filtered Events.
+     */
+    Events since(final LocalDateTime timestamp);
+
+    /**
+     * Show events created until the specified timestamp,
+     * then stop streaming.
+     * @param timestamp Given timestamp.
+     * @return Filtered Events.
+     */
+    Events until(final LocalDateTime timestamp);
+
+    /**
+     * Filter these Events.
+     * @param filter Supplier of filters.
+     * @return Filtered Events.
+     * @see <a href="https://docs.docker.com/engine/api/v1.40/#operation/SystemEvents">Docker API Docs</a>
+     */
+    Events filter(final Supplier<Map<String, Iterable<String>>> filter);
+
+    /**
+     * Start monitoring these events. Pay attention:<br><br>
+     * The Stream is <b>infinite</b>, which means you have to specify a
+     * <b>limit</b> before calling a terminal operation on it. Otherwise,
+     * your terminal operation will run until the Server closes the connection.
+     * Example:
+     * <pre>
+     *   final Docker docker = ...;
+     *   final List&lt;JsonObject&gt; firstTen = docker.events().monitor()
+     *       .limit(10).collect(Collectors.toList());
+     *   //It will wait for and return the first 10 real-time events
+     *   //from the server.
+     * </pre>
+     * @throws IOException If there is any I/O problem.
+     * @throws UnexpectedResponseException If the response is not 200 OK.
+     * @return Stream of events.
+     */
+    Stream<JsonObject> monitor()
+        throws IOException, UnexpectedResponseException;
+
+    /**
+     * Docker where these events came from.
+     * @return Docker.
+     */
+    Docker docker();
+}

src/main/java/com/amihaiemil/docker/RtDocker.java

@@ -25,23 +25,13 @@
  */
 package com.amihaiemil.docker;
 
-import java.io.ByteArrayInputStream;
 import java.io.IOException;
-import java.io.InputStream;
 import java.net.URI;
-import java.util.stream.Stream;
-
-
 import org.apache.http.HttpResponse;
 import org.apache.http.HttpStatus;
 import org.apache.http.client.HttpClient;
-import org.apache.http.client.methods.CloseableHttpResponse;
 import org.apache.http.client.methods.HttpGet;
 
-import javax.json.Json;
-import javax.json.JsonObject;
-import javax.json.JsonReader;
-
 /**
  * Restful Docker.
  * @author Mihai Andronache (amihaiemil@gmail.com)
@@ -78,73 +68,13 @@ public final boolean ping() throws IOException {
         return response.getStatusLine().getStatusCode() == HttpStatus.SC_OK;
     }
 
-    /**
-     * Unlike other methods, we cannot implement this one using Response
-     * Handlers, because Apache HTTP Client tries to consume the remaining
-     * content after all the handlers have been executed, which results in
-     * a blockage, since the underlying InputStream is potentially infinite.
-     *
-     * @checkstyle CommentsIndentation (100 lines)
-     * @checkstyle Indentation (100 lines)
-     * @return Stream of Events.
-     * @throws IOException If any I/O problem occurs.
-     * @throws UnexpectedResponseException If the response status is not 200.
-     */
     @Override
-    public Stream<JsonObject> events()
-        throws IOException, UnexpectedResponseException {
-        final HttpGet monitor = new HttpGet(
-            this.baseUri.toString() + "/events"
+    public Events events() {
+        return new RtEvents(
+            this.client,
+            URI.create(this.baseUri.toString() + "/events"),
+            this
         );
-        final HttpResponse response = this.client.execute(monitor);
-        final int actual = response.getStatusLine().getStatusCode();
-        if(actual != HttpStatus.SC_OK) {
-            throw new UnexpectedResponseException(
-                this.baseUri.toString() + "/events",
-                actual, HttpStatus.SC_OK,
-                Json.createObjectBuilder().build()
-            );
-        } else {
-            final InputStream content = response.getEntity().getContent();
-            final Stream<JsonObject> stream = Stream.generate(
-                () -> {
-                    JsonObject read = null;
-                    try {
-                        final byte[] tmp = new byte[4096];
-                        while (content.read(tmp) != -1) {
-                            try {
-                                final JsonReader reader = Json.createReader(
-                                    new ByteArrayInputStream(tmp)
-                                );
-                                read = reader.readObject();
-                                break;
-                            //@checkstyle IllegalCatch (1 line)
-                            } catch (final Exception exception) {
-                                //Couldn't parse byte[] to Json,
-                                //try to read more bytes.
-                            }
-                        }
-                    } catch (final IOException ex) {
-                        throw new IllegalStateException(
-                            "IOException when reading streamed JsonObjects!"
-                        );
-                    }
-                    return read;
-                }
-            ).onClose(
-                () -> {
-                    try {
-                        ((CloseableHttpResponse) response).close();
-                    } catch (final IOException ex) {
-                        //There is a bug in Apache HTTPClient, when closing
-                        //an infinite InputStream: IOException is thrown
-                        //because the client still tries to read the remainder
-                        // of the closed Stream. We should ignore this case.
-                    }
-                }
-            );
-            return stream;
-        }
     }
 
     @Override