Log sorting

Logs from the Doppler log streaming endpoint are not guaranteed to be in order
temporally.  Previously, the log streaming operation (Applications#logs)
punted on this issue, requiring the user to deal with it.  Thanks to Kris De
Volder doing the hard work of designing a solid reactive algorithm, we can now
include this functionality in the client itself.

Author: nebhale

.idea/dictionaries/bhale.xml

@@ -92,7 +92,7 @@ private String getTokenUri(String root) {
             .queryParam("grant_type", "password")
             .queryParam("username", getUsername())
             .queryParam("password", getPassword())
-            .build().toUriString();
+            .build().encode().toUriString();
     }
 
 }

cloudfoundry-client-reactor/src/main/java/org/cloudfoundry/reactor/tokenprovider/_PasswordGrantTokenProvider.java

@@ -91,6 +91,7 @@
 import org.cloudfoundry.util.OperationUtils;
 import org.cloudfoundry.util.PaginationUtils;
 import org.cloudfoundry.util.ResourceUtils;
+import org.cloudfoundry.util.SortingUtils;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import reactor.util.function.Tuple2;
@@ -100,6 +101,7 @@
 import java.io.InputStream;
 import java.time.Duration;
 import java.util.Collections;
+import java.util.Comparator;
 import java.util.Date;
 import java.util.HashMap;
 import java.util.List;
@@ -122,6 +124,10 @@ public final class DefaultApplications implements Applications {
 
     private static final int CF_STAGING_NOT_FINISHED = 170002;
 
+    private static final Comparator<LogMessage> LOG_MESSAGE_COMPARATOR = (a, b) -> a.getTimestamp().compareTo(b.getTimestamp());
+
+    private static final Duration LOG_MESSAGE_TIMESPAN = Duration.ofMillis(500);
+
     private static final int MAX_NUMBER_OF_RECENT_EVENTS = 50;
 
     private static final String STARTED_STATE = "STARTED";
@@ -654,12 +660,13 @@ private static Flux<LogMessage> getLogs(Mono<DopplerClient> dopplerClient, Strin
             return requestLogsRecent(dopplerClient, applicationId)
                 .filter(e -> EventType.LOG_MESSAGE == e.getEventType())
                 .map(Envelope::getLogMessage)
-                .collectSortedList((a, b) -> a.getTimestamp().compareTo(b.getTimestamp()))
+                .collectSortedList(LOG_MESSAGE_COMPARATOR)
                 .flatMapIterable(d -> d);
         } else {
             return requestLogsStream(dopplerClient, applicationId)
                 .filter(e -> EventType.LOG_MESSAGE == e.getEventType())
-                .map(Envelope::getLogMessage);
+                .map(Envelope::getLogMessage)
+                .compose(SortingUtils.timespan(LOG_MESSAGE_COMPARATOR, LOG_MESSAGE_TIMESPAN));
         }
     }
 

cloudfoundry-operations/src/main/java/org/cloudfoundry/operations/applications/DefaultApplications.java

@@ -2475,7 +2475,6 @@ public static final class LogsRecentNotSet extends AbstractOperationsApiTest<Log
 
         private final DefaultApplications applications = new DefaultApplications(Mono.just(this.cloudFoundryClient), Mono.just(this.dopplerClient), Mono.just(TEST_SPACE_ID));
 
-
         @Before
         public void setUp() throws Exception {
             requestApplications(this.cloudFoundryClient, "test-application-name", TEST_SPACE_ID, "test-metadata-id");

cloudfoundry-operations/src/test/java/org/cloudfoundry/operations/applications/DefaultApplicationsTest.java

@@ -20,6 +20,9 @@
 import org.cloudfoundry.client.v2.Resource;
 import reactor.core.publisher.Flux;
 
+/**
+ * Utilities for dealing with {@link Resource}s
+ */
 public final class ResourceUtils {
 
     private ResourceUtils() {

cloudfoundry-util/src/main/java/org/cloudfoundry/util/ResourceUtils.java

@@ -0,0 +1,89 @@
+/*
+ * Copyright 2013-2016 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.cloudfoundry.util;
+
+import reactor.core.Cancellation;
+import reactor.core.publisher.DirectProcessor;
+import reactor.core.publisher.Flux;
+import reactor.util.function.Tuple2;
+
+import java.time.Duration;
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.List;
+import java.util.PriorityQueue;
+import java.util.Queue;
+import java.util.function.Function;
+
+/**
+ * Utilities for sorting
+ */
+public final class SortingUtils {
+
+    private SortingUtils() {
+    }
+
+    /**
+     * Sorts the elements of a {@link Flux} within a sliding time window.  This sorter should be used when element order may be scrambled, but that scrambling has a certain 'temporal locality' to it.
+     * This assumption means that sorting can be limited to elements that arrive temporally close to one another without risking a latecomer being sorted incorrectly.
+     *
+     * @param comparator a {@link Comparator} to use when sorting the elements within the window
+     * @param timespan   the duration of the 'temporal locality'
+     * @param <T>        The type of the elements to be sorted
+     * @return a {@link Flux} providing the sorted elements
+     */
+    public static <T> Function<Flux<T>, Flux<T>> timespan(Comparator<T> comparator, Duration timespan) {
+        return source -> {
+            Queue<Tuple2<Long, T>> accumulator = new PriorityQueue<>((o1, o2) -> comparator.compare(o1.t2, o2.t2));
+            Object monitor = new Object();
+
+            DirectProcessor<Void> d = DirectProcessor.create();
+
+            Cancellation cancellation = source
+                .timestamp()
+                .subscribe(e -> {
+                    synchronized (monitor) {
+                        accumulator.add(e);
+                    }
+                }, d::onError, d::onComplete);
+
+            return Flux
+                .interval(timespan)
+                .takeUntilOther(d)
+                .flatMap(n -> getItems(accumulator, comparator, timespan), null, () -> getItems(accumulator, comparator, Duration.ZERO))
+                .doOnCancel(cancellation::dispose);
+        };
+    }
+
+    private static <T> Flux<T> getItems(Queue<Tuple2<Long, T>> accumulator, Object monitor, Duration timespan) {
+        List<T> items = new ArrayList<>();
+
+        synchronized (monitor) {
+            while (isBefore(accumulator.peek(), timespan)) {
+                items.add(accumulator.remove().t2);
+            }
+        }
+
+        return Flux.fromIterable(items);
+    }
+
+    private static <T> boolean isBefore(Tuple2<Long, T> candidate, Duration timespan) {
+        return candidate != null && (Duration.ZERO == timespan || Instant.ofEpochMilli(candidate.t1).isBefore(Instant.now().minus(timespan)));
+    }
+
+}