Sending implements SendingOrDropped { @SuppressWarnings("rawtypes") @Override public Optional<CompletionStage<CommandResponse>> monitorAndAcknowledge( final ExceptionToAcknowledgementConverter exceptionConverter) { return Optional.of(futureResponse .thenApply(response -> acknowledge(response, exceptionConverter)) .thenApply(this::monitor) .exceptionally(error -> { final Acknowledgement result = handleException(getRootCause(error), exceptionConverter); monitor(result); return result; })); } Sending(final SendingContext sendingContext, final CompletionStage<CommandResponse<?>> futureResponse,
            final ThreadSafeDittoLoggingAdapter logger); @SuppressWarnings("rawtypes") @Override Optional<CompletionStage<CommandResponse>> monitorAndAcknowledge(
            final ExceptionToAcknowledgementConverter exceptionConverter);  }
@Test public void acknowledgeAndMonitorNullCommandResponseWhenShouldAcknowledge() { Mockito.when(autoAckTarget.getIssuedAcknowledgementLabel()).thenReturn(Optional.of(ACKNOWLEDGEMENT_LABEL)); final var source = Mockito.mock(Signal.class); final var thingId = ThingId.generateRandom(); Mockito.when(source.getEntityId()).thenReturn(thingId); Mockito.when(source.getDittoHeaders()).thenReturn(dittoHeaders); Mockito.when(mappedOutboundSignal.getSource()).thenReturn(source); final Acknowledgement expectedResponse = exceptionConverter.convertException(MessageSendingFailedException.newBuilder() .message("Message sending terminated without the expected acknowledgement.") .description("Please contact the service team.") .build(), ACKNOWLEDGEMENT_LABEL, thingId, dittoHeaders); final var expectedException = MessageSendingFailedException.newBuilder() .message("Message sending terminated without the expected acknowledgement.") .description("Please contact the service team.") .build(); final Sending underTest = new Sending(sendingContext, CompletableFuture.completedStage(null), logger); final Optional<CompletionStage<CommandResponse>> result = underTest.monitorAndAcknowledge(exceptionConverter); Mockito.verify(publishedMonitor).failure(externalMessage, expectedException); Mockito.verify(acknowledgedMonitor) .failure(Mockito.eq(externalMessage), Mockito.any(DittoRuntimeException.class)); assertThat(result).hasValueSatisfying( resultFuture -> assertThat(resultFuture).isCompletedWithValue(expectedResponse)); }
@Test public void monitorAcknowledgementSendSuccessKeepOriginalResponse() { final var acknowledgement = Mockito.mock(Acknowledgement.class); final var acknowledgementStatusCode = HttpStatusCode.ACCEPTED; Mockito.when(acknowledgement.getStatusCode()).thenReturn(acknowledgementStatusCode); final Sending underTest = new Sending(sendingContext, CompletableFuture.completedStage(acknowledgement), logger); final Optional<CompletionStage<CommandResponse>> result = underTest.monitorAndAcknowledge(exceptionConverter); Mockito.verifyNoInteractions(acknowledgedMonitor); Mockito.verify(publishedMonitor).success(externalMessage); assertThat(result) .hasValueSatisfying(resultFuture -> assertThat(resultFuture).isCompletedWithValue(acknowledgement)); }
@Test public void monitorAcknowledgementSendFailureKeepOriginalResponse() { final var acknowledgement = Mockito.mock(Acknowledgement.class); final var acknowledgementPayload = JsonObject.newBuilder().set("foo", "bar").build(); final var acknowledgementStatusCode = HttpStatusCode.INTERNAL_SERVER_ERROR; Mockito.when(acknowledgement.getStatusCode()).thenReturn(acknowledgementStatusCode); Mockito.when(acknowledgement.getEntity()).thenReturn(Optional.of(acknowledgementPayload)); final var expectedException = MessageSendingFailedException.newBuilder() .statusCode(acknowledgementStatusCode) .message("Received negative acknowledgement for label <" + ACKNOWLEDGEMENT_LABEL + ">.") .description("Payload: " + acknowledgementPayload) .build(); final Sending underTest = new Sending(sendingContext, CompletableFuture.completedStage(acknowledgement), logger); final Optional<CompletionStage<CommandResponse>> result = underTest.monitorAndAcknowledge(exceptionConverter); Mockito.verifyNoInteractions(publishedMonitor); Mockito.verify(acknowledgedMonitor).failure(externalMessage, expectedException); assertThat(result) .hasValueSatisfying(resultFuture -> assertThat(resultFuture).isCompletedWithValue(acknowledgement)); }
@Test public void monitorLiveResponseSendSuccessKeepOriginalResponse() { final var issuedAckLabel = DittoAcknowledgementLabel.LIVE_RESPONSE; Mockito.when(autoAckTarget.getIssuedAcknowledgementLabel()).thenReturn(Optional.of(issuedAckLabel)); final CommandResponse<?> commandResponse = Mockito.mock(CommandResponse.class); final var commandResponseStatusCode = HttpStatusCode.ACCEPTED; Mockito.when(commandResponse.getStatusCode()).thenReturn(commandResponseStatusCode); final Sending underTest = new Sending(sendingContext, CompletableFuture.completedStage(commandResponse), logger); final Optional<CompletionStage<CommandResponse>> result = underTest.monitorAndAcknowledge(exceptionConverter); Mockito.verifyNoInteractions(acknowledgedMonitor); Mockito.verify(publishedMonitor).success(externalMessage); assertThat(result) .hasValueSatisfying(resultFuture -> assertThat(resultFuture).isCompletedWithValue(commandResponse)); }
@Test public void monitorLiveResponseSendFailureKeepOriginalResponse() { final var issuedAckLabel = DittoAcknowledgementLabel.LIVE_RESPONSE; Mockito.when(autoAckTarget.getIssuedAcknowledgementLabel()).thenReturn(Optional.of(issuedAckLabel)); final CommandResponse<?> commandResponse = Mockito.mock(CommandResponse.class); final var commandResponsePayload = JsonObject.newBuilder().set("foo", "bar").build(); final var commandResponseStatusCode = HttpStatusCode.CONFLICT; Mockito.when(commandResponse.getStatusCode()).thenReturn(commandResponseStatusCode); Mockito.when(commandResponse.toJson()).thenReturn(commandResponsePayload); final var expectedException = MessageSendingFailedException.newBuilder() .statusCode(commandResponseStatusCode) .message("Received negative acknowledgement for label <" + issuedAckLabel + ">.") .description("Payload: " + commandResponsePayload) .build(); final Sending underTest = new Sending(sendingContext, CompletableFuture.completedStage(commandResponse), logger); final Optional<CompletionStage<CommandResponse>> result = underTest.monitorAndAcknowledge(exceptionConverter); Mockito.verifyNoInteractions(publishedMonitor); Mockito.verify(acknowledgedMonitor).failure(externalMessage, expectedException); assertThat(result) .hasValueSatisfying(resultFuture -> assertThat(resultFuture).isCompletedWithValue(commandResponse)); }
@Test public void monitorNothingAsShouldNotAcknowledgeAndOriginalResponseIsNull() { sendingContext = SendingContext.newBuilder() .mappedOutboundSignal(mappedOutboundSignal) .externalMessage(externalMessage) .genericTarget(genericTarget) .publishedMonitor(publishedMonitor) .droppedMonitor(droppedMonitor) .build(); final CommandResponse<?> commandResponse = null; final Sending underTest = new Sending(sendingContext, CompletableFuture.completedStage(commandResponse), logger); final Optional<CompletionStage<CommandResponse>> result = underTest.monitorAndAcknowledge(exceptionConverter); Mockito.verifyNoInteractions(publishedMonitor, acknowledgedMonitor); assertThat(result) .hasValueSatisfying(resultFuture -> assertThat(resultFuture).isCompletedWithValue(commandResponse)); }
@Test public void monitorAndAcknowledgeWhenFutureResponseTerminatedExceptionallyAndNoAckLabelIssued() { final var rootCause = new IllegalStateException("A foo must not bar! Never ever!"); final Sending underTest = new Sending(sendingContext, CompletableFuture.failedStage(rootCause), logger); final Optional<CompletionStage<CommandResponse>> result = underTest.monitorAndAcknowledge(exceptionConverter); Mockito.verifyNoInteractions(acknowledgedMonitor); Mockito.verify(publishedMonitor).exception(Mockito.eq(externalMessage), Mockito.eq(rootCause)); Mockito.verify(logger).withCorrelationId(testName.getMethodName()); assertThat(result).hasValueSatisfying(resultFuture -> assertThat(resultFuture).isCompletedWithValue(null)); }