ConsumerMessageSender { public void sendAsync(Message message) { sendAsync(message, calculateMessageDelay(message.getPublishingTimestamp())); } ConsumerMessageSender(Subscription subscription,
                                 MessageSenderFactory messageSenderFactory,
                                 List<SuccessHandler> successHandlers,
                                 List<ErrorHandler> errorHandlers,
                                 SerialConsumerRateLimiter rateLimiter,
                                 ExecutorService deliveryReportingExecutor,
                                 InflightsPool inflight,
                                 HermesMetrics hermesMetrics,
                                 int asyncTimeoutMs,
                                 FutureAsyncTimeout<MessageSendingResult> futureAsyncTimeout,
                                 Clock clock); void initialize(); void shutdown(); void sendAsync(Message message); void updateSubscription(Subscription newSubscription);  }
@Test public void shouldHandleSuccessfulSending() { Message message = message(); when(messageSender.send(message)).thenReturn(success()); sender.sendAsync(message); verify(successHandler, timeout(1000)).handleSuccess(eq(message), eq(subscription), any(MessageSendingResult.class)); verifySemaphoreReleased(); verifyRateLimiterSuccessfulSendingCountedTimes(1); verifyLatencyTimersCountedTimes(1, 1); verifyZeroInteractions(errorHandler); verifyZeroInteractions(failedMeter); }
@Test public void shouldKeepTryingToSendMessageFailedSending() throws InterruptedException { Message message = message(); doReturn(failure()).doReturn(failure()).doReturn(success()).when(messageSender).send(message); sender.sendAsync(message); verify(successHandler, timeout(1000)).handleSuccess(eq(message), eq(subscription), any(MessageSendingResult.class)); verifySemaphoreReleased(); verifyLatencyTimersCountedTimes(3, 3); verifyRateLimiterFailedSendingCountedTimes(2); verifyRateLimiterSuccessfulSendingCountedTimes(1); verifyErrorHandlerHandleFailed(message, subscription, 2); }
@Test public void shouldDiscardMessageWhenTTLIsExceeded() { Message message = messageWithTimestamp(System.currentTimeMillis() - 11000); doReturn(failure()).when(messageSender).send(message); sender.sendAsync(message); verify(errorHandler, timeout(1000)).handleDiscarded(eq(message), eq(subscription), any(MessageSendingResult.class)); verifySemaphoreReleased(); verifyZeroInteractions(successHandler); verifyLatencyTimersCountedTimes(1, 1); verifyRateLimiterFailedSendingCountedTimes(1); }
@Test public void shouldNotKeepTryingToSendMessageFailedWithStatusCode4xx() throws InterruptedException { Message message = message(); doReturn(failure(403)).doReturn(success()).when(messageSender).send(message); sender.sendAsync(message); verify(errorHandler, timeout(1000)).handleDiscarded(eq(message), eq(subscription), any(MessageSendingResult.class)); verifySemaphoreReleased(); verifyZeroInteractions(successHandler); verifyLatencyTimersCountedTimes(1, 1); }
@Test public void shouldTreat4xxResponseForSubscriptionWithNo4xxRetryAsSuccess() { Message message = message(); doReturn(failure(403)).doReturn(success()).when(messageSender).send(message); sender.sendAsync(message); verifyRateLimiterFailedSendingCountedTimes(0); verifyRateLimiterSuccessfulSendingCountedTimes(1); verifyErrorHandlerHandleFailed(message, subscription, 1); }
@Test public void shouldReduceSendingRateLimitOnErrorResponseOtherThan4xxForSubscriptionWithNo4xxRetry() { Message message = message(); doReturn(failure(500)).doReturn(success()).when(messageSender).send(message); sender.sendAsync(message); verifyRateLimiterFailedSendingCountedTimes(1); verifyErrorHandlerHandleFailed(message, subscription, 1); }
@Test public void shouldRaiseTimeoutWhenSenderNotCompletesResult() { Message message = message(); when(messageSender.send(message)).thenReturn(new CompletableFuture<>()); sender.sendAsync(message); verifyErrorHandlerHandleFailed(message, subscription, 1, 3000); }
@Test public void shouldNotBackoffRetriesOnRetryAfter() throws InterruptedException { int retrySeconds = 1; Message message = message(); doReturn(backoff(retrySeconds)).doReturn(success()).when(messageSender).send(message); sender.sendAsync(message); verifyRateLimiterFailedSendingCountedTimes(0); verifyRateLimiterSuccessfulSendingCountedTimes(2); verifySemaphoreReleased(); }
@Test public void shouldBackoffRetriesOnServiceUnavailableWithoutRetryAfter() throws InterruptedException { Message message = message(); doReturn(failure(SERVICE_UNAVAILABLE.code())).doReturn(success()).when(messageSender).send(message); sender.sendAsync(message); verifyRateLimiterFailedSendingCountedTimes(1); verifyRateLimiterSuccessfulSendingCountedTimes(1); verifySemaphoreReleased(); }
@Test public void shouldBackoffRetriesOnTooManyRequestsWithoutRetryAfter() throws InterruptedException { Message message = message(); doReturn(failure(TOO_MANY_REQUESTS.code())).doReturn(success()).when(messageSender).send(message); sender.sendAsync(message); verifyRateLimiterSuccessfulSendingCountedTimes(1); verifySemaphoreReleased(); }
@Test public void shouldNotRetryOnRetryAfterAboveTtl() throws InterruptedException { int retrySeconds = subscription.getSerialSubscriptionPolicy().getMessageTtl(); Message message = message(); doReturn(backoff(retrySeconds)).when(messageSender).send(message); sender.sendAsync(message); verifyRateLimiterSuccessfulSendingCountedTimes(1); verifyErrorHandlerHandleDiscarded(message, subscription); verifySemaphoreReleased(); }