Removing executor for HttpRequestServiceTest

Author: LikeTheSalad

opamp-client/src/main/java/io/opentelemetry/opamp/client/internal/request/service/HttpRequestService.java

@@ -94,7 +94,7 @@ public void start(Callback callback, Supplier<Request> requestSupplier) {
       this.requestSupplier = requestSupplier;
       executor.start(this);
     } else {
-      throw new IllegalStateException("RequestDispatcher is already running");
+      throw new IllegalStateException("HttpRequestService is already running");
     }
   }
 

opamp-client/src/test/java/io/opentelemetry/opamp/client/internal/request/service/HttpRequestServiceTest.java

@@ -10,13 +10,10 @@
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.anyInt;
 import static org.mockito.ArgumentMatchers.eq;
-import static org.mockito.Mockito.clearInvocations;
 import static org.mockito.Mockito.doThrow;
-import static org.mockito.Mockito.inOrder;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.verifyNoInteractions;
 import static org.mockito.Mockito.when;
 
 import io.opentelemetry.opamp.client.internal.connectivity.http.HttpErrorException;
@@ -28,82 +25,64 @@
 import io.opentelemetry.opamp.client.internal.request.delay.PeriodicTaskExecutor;
 import io.opentelemetry.opamp.client.internal.response.Response;
 import java.io.ByteArrayInputStream;
+import java.io.Closeable;
 import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.NoSuchElementException;
+import java.util.Queue;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.Supplier;
 import opamp.proto.AgentToServer;
 import opamp.proto.RetryInfo;
 import opamp.proto.ServerErrorResponse;
 import opamp.proto.ServerErrorResponseType;
 import opamp.proto.ServerToAgent;
+import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
 import org.mockito.ArgumentCaptor;
-import org.mockito.InOrder;
 import org.mockito.Mock;
 import org.mockito.junit.jupiter.MockitoExtension;
 
 @SuppressWarnings("unchecked")
 @ExtendWith(MockitoExtension.class)
 class HttpRequestServiceTest {
-  @Mock private HttpSender requestSender;
-  @Mock private PeriodicDelay periodicRequestDelay;
-  @Mock private TestPeriodicRetryDelay periodicRetryDelay;
-  @Mock private PeriodicTaskExecutor executor;
   @Mock private RequestService.Callback callback;
   @Mock private Supplier<Request> requestSupplier;
+  private TestHttpSender requestSender;
+  private final PeriodicDelay periodicRequestDelay =
+      PeriodicDelay.ofFixedDuration(Duration.ofSeconds(1));
+  private TestPeriodicRetryDelay periodicRetryDelay;
   private int requestSize = -1;
   private HttpRequestService httpRequestService;
 
   @BeforeEach
   void setUp() {
+    requestSender = new TestHttpSender();
+    periodicRetryDelay = new TestPeriodicRetryDelay(Duration.ofSeconds(2));
     httpRequestService =
         new HttpRequestService(
             requestSender,
-            executor,
+            PeriodicTaskExecutor.create(periodicRequestDelay),
             periodicRequestDelay,
             periodicRetryDelay,
             RetryAfterParser.getInstance());
-  }
-
-  @Test
-  void verifyStart() {
     httpRequestService.start(callback, requestSupplier);
-
-    InOrder inOrder = inOrder(periodicRequestDelay, executor);
-    inOrder.verify(executor).start(httpRequestService);
-
-    // Try starting it again:
-    try {
-      httpRequestService.start(callback, requestSupplier);
-      fail();
-    } catch (IllegalStateException e) {
-      assertThat(e).hasMessage("RequestDispatcher is already running");
-    }
   }
 
-  @Test
-  void verifyStop() {
-    httpRequestService.start(callback, requestSupplier);
-    httpRequestService.stop();
-
-    verify(executor).stop();
-
-    // Try stopping it again:
-    clearInvocations(executor);
-    httpRequestService.stop();
-    verifyNoInteractions(executor);
-  }
-
-  @Test
-  void verifyStop_whenNotStarted() {
+  @AfterEach
+  void tearDown() {
+    requestSender.close();
     httpRequestService.stop();
-
-    verifyNoInteractions(executor, requestSender, periodicRequestDelay);
   }
 
   @Test
@@ -126,9 +105,12 @@ void verifySendingRequest_happyPath() {
     prepareRequest();
     enqueueResponse(httpResponse);
 
-    httpRequestService.run();
+    httpRequestService.sendRequest();
 
-    verify(requestSender).send(any(), eq(requestSize));
+    requestSender.awaitForRequest(Duration.ofMillis(500));
+    assertThat(requestSender.requests).hasSize(1);
+    assertThat(requestSender.requests.get(0).contentLength).isEqualTo(requestSize);
+    verify(callback).onConnectionSuccess();
     verify(callback).onRequestSuccess(Response.create(serverToAgent));
   }
 
@@ -186,7 +168,6 @@ void verifySendingRequest_whenThereIsAGenericHttpError() {
     httpRequestService.run();
 
     verifyRequestFailedCallback(500);
-    verifyNoInteractions(executor);
   }
 
   @Test
@@ -200,7 +181,7 @@ void verifySendingRequest_whenThereIsATooManyRequestsError() {
     httpRequestService.run();
 
     verifyRequestFailedCallback(429);
-    verify(executor).setPeriodicDelay(periodicRetryDelay);
+    //    verify(executor).setPeriodicDelay(periodicRetryDelay); todo
     verify(periodicRetryDelay, never()).suggestDelay(any());
   }
 
@@ -216,7 +197,7 @@ void verifySendingRequest_whenThereIsATooManyRequestsError_withSuggestedDelay()
     httpRequestService.run();
 
     verifyRequestFailedCallback(429);
-    verify(executor).setPeriodicDelay(periodicRetryDelay);
+    //    verify(executor).setPeriodicDelay(periodicRetryDelay); todo
     verify(periodicRetryDelay).suggestDelay(Duration.ofSeconds(5));
   }
 
@@ -239,7 +220,7 @@ void verifySendingRequest_whenServerProvidesRetryInfo_usingTheProvidedInfo() {
 
     verify(callback).onRequestSuccess(Response.create(serverToAgent));
     verify(periodicRetryDelay).suggestDelay(Duration.ofNanos(nanosecondsToWaitForRetry));
-    verify(executor).setPeriodicDelay(periodicRetryDelay);
+    //    verify(executor).setPeriodicDelay(periodicRetryDelay); todo
   }
 
   @Test
@@ -258,7 +239,7 @@ void verifySendingRequest_whenServerIsUnavailable() {
 
     verify(callback).onRequestSuccess(Response.create(serverToAgent));
     verify(periodicRetryDelay, never()).suggestDelay(any());
-    verify(executor).setPeriodicDelay(periodicRetryDelay);
+    //    verify(executor).setPeriodicDelay(periodicRetryDelay); todo
   }
 
   @Test
@@ -272,7 +253,7 @@ void verifySendingRequest_whenThereIsAServiceUnavailableError() {
     httpRequestService.run();
 
     verifyRequestFailedCallback(503);
-    verify(executor).setPeriodicDelay(periodicRetryDelay);
+    //    verify(executor).setPeriodicDelay(periodicRetryDelay); todo
     verify(periodicRetryDelay, never()).suggestDelay(any());
   }
 
@@ -288,7 +269,7 @@ void verifySendingRequest_whenThereIsAServiceUnavailableError_withSuggestedDelay
     httpRequestService.run();
 
     verifyRequestFailedCallback(503);
-    verify(executor).setPeriodicDelay(periodicRetryDelay);
+    //    verify(executor).setPeriodicDelay(periodicRetryDelay); todo
     verify(periodicRetryDelay).suggestDelay(Duration.ofSeconds(2));
   }
 
@@ -303,7 +284,7 @@ private void verifyRequestFailedCallback(int errorCode) {
   void verifySendingRequest_duringRegularMode() {
     httpRequestService.sendRequest();
 
-    verify(executor).executeNow();
+    //    verify(executor).executeNow(); todo
   }
 
   @Test
@@ -312,7 +293,7 @@ void verifySendingRequest_duringRetryMode() {
 
     httpRequestService.sendRequest();
 
-    verify(executor, never()).executeNow();
+    //    verify(executor, never()).executeNow(); todo
   }
 
   @Test
@@ -325,7 +306,7 @@ void verifySuccessfulSendingRequest_duringRetryMode() {
 
     httpRequestService.run();
 
-    verify(executor).setPeriodicDelay(periodicRequestDelay);
+    //    verify(executor).setPeriodicDelay(periodicRequestDelay); todo
   }
 
   private void enableRetryMode() {
@@ -339,17 +320,14 @@ private void enableRetryMode() {
   }
 
   private void prepareRequest() {
-    httpRequestService.start(callback, requestSupplier);
-    clearInvocations(executor);
     AgentToServer agentToServer = new AgentToServer.Builder().sequence_num(10).build();
     requestSize = agentToServer.encodeByteString().size();
     Request request = Request.create(agentToServer);
     when(requestSupplier.get()).thenReturn(request);
   }
 
   private void enqueueResponse(HttpSender.Response httpResponse) {
-    when(requestSender.send(any(), anyInt()))
-        .thenReturn(CompletableFuture.completedFuture(httpResponse));
+    requestSender.enqueueResponse(httpResponse);
   }
 
   private static void attachServerToAgentMessage(
@@ -360,16 +338,79 @@ private static void attachServerToAgentMessage(
   }
 
   private static class TestPeriodicRetryDelay implements PeriodicDelay, AcceptsDelaySuggestion {
+    private final Duration delay;
+
+    private TestPeriodicRetryDelay(Duration delay) {
+      this.delay = delay;
+    }
 
     @Override
     public void suggestDelay(Duration delay) {}
 
     @Override
     public Duration getNextDelay() {
-      return null;
+      return delay;
     }
 
     @Override
     public void reset() {}
   }
+
+  private static class TestHttpSender implements HttpSender, Closeable {
+    private final List<RequestParams> requests = Collections.synchronizedList(new ArrayList<>());
+    private final Queue<HttpSender.Response> responses = new ConcurrentLinkedQueue<>();
+    private final AtomicInteger unexpectedRequests = new AtomicInteger(0);
+    private volatile CountDownLatch latch;
+
+    @Override
+    public CompletableFuture<HttpSender.Response> send(BodyWriter writer, int contentLength) {
+      requests.add(new RequestParams(contentLength));
+      HttpSender.Response response = null;
+      try {
+        response = responses.remove();
+        if (latch != null) {
+          latch.countDown();
+        }
+      } catch (NoSuchElementException e) {
+        unexpectedRequests.incrementAndGet();
+      }
+      return CompletableFuture.completedFuture(response);
+    }
+
+    public void enqueueResponse(HttpSender.Response response) {
+      responses.add(response);
+    }
+
+    public void awaitForRequest(Duration timeout) {
+      if (latch != null) {
+        throw new IllegalStateException();
+      }
+      latch = new CountDownLatch(1);
+      try {
+        if (!latch.await(timeout.toMillis(), TimeUnit.MILLISECONDS)) {
+          fail("No request received before timeout " + timeout);
+        }
+      } catch (InterruptedException e) {
+        throw new RuntimeException(e);
+      } finally {
+        latch = null;
+      }
+    }
+
+    @Override
+    public void close() {
+      int count = unexpectedRequests.get();
+      if (count > 0) {
+        fail("Unexpected requests count: " + count);
+      }
+    }
+
+    private static class RequestParams {
+      public final int contentLength;
+
+      private RequestParams(int contentLength) {
+        this.contentLength = contentLength;
+      }
+    }
+  }
 }