[test-only] Simulate dialogue clients against a DeterministicScheduler

.gitignore

@@ -56,5 +56,3 @@ __init__.pyc
 .cache/
 .ipynb_checkpoints/
 .vscode/
-
-

dialogue-core/build.gradle

@@ -15,6 +15,7 @@ dependencies {
     testCompile 'junit:junit'
     testCompile 'org.assertj:assertj-core'
     testCompile 'org.mockito:mockito-core'
+    testRuntimeOnly 'org.slf4j:slf4j-simple'
 
     annotationProcessor 'org.immutables:value'
     compile 'org.immutables:value::annotations'

dialogue-core/src/main/java/com/palantir/dialogue/core/ConcurrencyLimitedChannel.java

@@ -42,22 +42,24 @@
  */
 final class ConcurrencyLimitedChannel implements LimitedChannel {
     private static final Void NO_CONTEXT = null;
+    private static final Supplier<Long> SYSTEM_NANOTIME = System::nanoTime;
 
     private final Channel delegate;
-    private final LoadingCache<Endpoint, Limiter<Void>> limiters;
+    private final LoadingCache<Endpoint, SimpleLimiter<Void>> limiters;
 
     @VisibleForTesting
-    ConcurrencyLimitedChannel(Channel delegate, Supplier<Limiter<Void>> limiterSupplier) {
+    ConcurrencyLimitedChannel(Channel delegate, Supplier<SimpleLimiter<Void>> limiterSupplier) {
         this.delegate = delegate;
         this.limiters =
                 Caffeine.newBuilder().expireAfterAccess(Duration.ofMinutes(5)).build(key -> limiterSupplier.get());
     }
 
     static ConcurrencyLimitedChannel create(Channel delegate) {
-        return new ConcurrencyLimitedChannel(delegate, ConcurrencyLimitedChannel::createLimiter);
+        return new ConcurrencyLimitedChannel(delegate, () -> ConcurrencyLimitedChannel.createLimiter(SYSTEM_NANOTIME));
     }
 
-    private static Limiter<Void> createLimiter() {
+    @VisibleForTesting
+    static SimpleLimiter<Void> createLimiter(Supplier<Long> nanoTimeClock) {
         AIMDLimit aimdLimit = AIMDLimit.newBuilder()
                 // Explicitly set values to prevent library changes from breaking us
                 .initialLimit(20)
@@ -68,13 +70,15 @@ private static Limiter<Void> createLimiter() {
                 .timeout(Long.MAX_VALUE, TimeUnit.DAYS)
                 .build();
         return SimpleLimiter.newBuilder()
+                .clock(nanoTimeClock)
                 .limit(WindowedLimit.newBuilder().build(aimdLimit))
                 .build();
     }
 
     @Override
     public Optional<ListenableFuture<Response>> maybeExecute(Endpoint endpoint, Request request) {
-        return limiters.get(endpoint).acquire(NO_CONTEXT).map(listener ->
+        SimpleLimiter<Void> limiter = limiters.get(endpoint);
+        return limiter.acquire(NO_CONTEXT).map(listener ->
                 DialogueFutures.addDirectCallback(delegate.execute(endpoint, request), new LimiterCallback(listener)));
     }
 

dialogue-core/src/main/java/com/palantir/dialogue/core/GenericRefreshingChannel.java

@@ -0,0 +1,65 @@
+/*
+ * (c) Copyright 2020 Palantir Technologies Inc. All rights reserved.
+ *
+ * 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 com.palantir.dialogue.core;
+
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
+import com.palantir.conjure.java.api.config.service.ServicesConfigBlock;
+import com.palantir.dialogue.Channel;
+import com.palantir.dialogue.Endpoint;
+import com.palantir.dialogue.Request;
+import com.palantir.dialogue.Response;
+import com.palantir.logsafe.exceptions.SafeIllegalStateException;
+import java.util.Optional;
+import java.util.function.Supplier;
+
+final class GenericRefreshingChannel implements Channel {
+    private final Supplier<Channel> channelSupplier;
+
+    private GenericRefreshingChannel(Supplier<Channel> channelSupplier) {
+        this.channelSupplier = channelSupplier;
+    }
+
+    /**
+     * Returns a refreshing {@link Channel} for a service identified by {@code service} in
+     * {@link ServicesConfigBlock#services()}.
+     */
+    public static <T> Channel create(Supplier<T> confSupplier, ChannelFactory<T> channelFactory) {
+        Supplier<Channel> channelSupplier = new MemoizingComposingSupplier<>(confSupplier, conf -> {
+            return channelFactory.create(conf).orElse(AlwaysThrowingChannel.INSTANCE);
+        });
+        return new GenericRefreshingChannel(channelSupplier);
+    }
+
+    @Override
+    public ListenableFuture<Response> execute(Endpoint endpoint, Request request) {
+        return channelSupplier.get().execute(endpoint, request);
+    }
+
+    private enum AlwaysThrowingChannel implements Channel {
+        INSTANCE;
+
+        @Override
+        public ListenableFuture<Response> execute(Endpoint _endpoint, Request _request) {
+            return Futures.immediateFailedFuture(new SafeIllegalStateException("Service not configured"));
+        }
+    }
+
+    public interface ChannelFactory<T> {
+        Optional<Channel> create(T conf);
+    }
+}

dialogue-core/src/main/java/com/palantir/dialogue/core/QueuedChannel.java

@@ -27,6 +27,7 @@
 import com.palantir.dialogue.Endpoint;
 import com.palantir.dialogue.Request;
 import com.palantir.dialogue.Response;
+import com.palantir.logsafe.SafeArg;
 import java.io.ByteArrayInputStream;
 import java.io.InputStream;
 import java.util.List;
@@ -37,6 +38,8 @@
 import java.util.concurrent.LinkedBlockingDeque;
 import java.util.concurrent.atomic.AtomicInteger;
 import org.immutables.value.Value;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
  * A {@link Channel} that queues requests while the underlying {@link LimitedChannel} is unable to accept any new
@@ -57,7 +60,7 @@
  * TODO(jellis): record metrics for queue sizes, num requests in flight, time spent in queue, etc.
  */
 final class QueuedChannel implements Channel {
-
+    private static final Logger log = LoggerFactory.getLogger(QueuedChannel.class);
     private static final Executor DIRECT = MoreExecutors.directExecutor();
 
     private final BlockingDeque<DeferredCall> queuedCalls;
@@ -114,8 +117,12 @@ private void onCompletion() {
      * Try to schedule as many tasks as possible. Called when requests are submitted and when they complete.
      */
     private void schedule() {
+        int numScheduled = 0;
         while (scheduleNextTask()) {
-            // Do nothing
+            numScheduled++;
+        }
+        if (numScheduled > 1) {
+            log.debug("Scheduled {} at the same time", SafeArg.of("numScheduled", numScheduled));
         }
     }
 

dialogue-core/src/test/java/com/palantir/dialogue/core/ConcurrencyLimitedChannelTest.java

@@ -23,6 +23,7 @@
 
 import com.google.common.util.concurrent.SettableFuture;
 import com.netflix.concurrency.limits.Limiter;
+import com.netflix.concurrency.limits.limiter.SimpleLimiter;
 import com.palantir.dialogue.Channel;
 import com.palantir.dialogue.Endpoint;
 import com.palantir.dialogue.Request;
@@ -47,7 +48,7 @@ public class ConcurrencyLimitedChannelTest {
     private Channel delegate;
 
     @Mock
-    private Limiter<Void> limiter;
+    private SimpleLimiter<Void> limiter;
 
     @Mock
     private Limiter.Listener listener;

settings.gradle

@@ -6,3 +6,4 @@ include 'dialogue-java-client'
 include 'dialogue-okhttp-client'
 include 'dialogue-serde'
 include 'dialogue-target'
+include 'simulation'

simulation/build.gradle

@@ -0,0 +1,24 @@
+versionsLock {
+  testProject()
+}
+
+dependencies {
+    implementation project(':dialogue-core')
+    implementation project(':dialogue-target')
+    implementation 'org.jmock:jmock'
+    implementation 'org.knowm.xchart:xchart'
+
+    testImplementation 'com.palantir.safe-logging:preconditions-assertj'
+    testImplementation 'junit:junit'
+    testImplementation 'org.assertj:assertj-core'
+    testImplementation 'org.mockito:mockito-core'
+
+    testRuntimeOnly 'org.slf4j:slf4j-simple'
+
+    annotationProcessor 'org.immutables:value'
+    compile 'org.immutables:value::annotations'
+}
+
+tasks.withType(JavaCompile) {
+    options.errorprone.errorproneArgs += '-Xep:Slf4jLogsafeArgs:OFF'
+}