RequestResponseEndpoint extends AbstractAmqpEndpoint<T> { protected final void handleRequestMessage( final ProtonConnection con, final ProtonReceiver receiver, final ResourceIdentifier targetAddress, final ProtonDelivery delivery, final Message requestMessage) { final HonoUser clientPrincipal = Constants.getClientPrincipal(con); final String replyTo = requestMessage.getReplyTo(); final SpanContext spanContext = TracingHelper.extractSpanContext(tracer, requestMessage); final Span currentSpan = TracingHelper.buildServerChildSpan(tracer, spanContext, "process request message", getName()) .withTag(Tags.HTTP_METHOD.getKey(), requestMessage.getSubject()) .withTag(Tags.MESSAGE_BUS_DESTINATION.getKey(), targetAddress.toString()) .start(); if (!passesFormalVerification(targetAddress, requestMessage)) { MessageHelper.rejected(delivery, new ErrorCondition(Constants.AMQP_BAD_REQUEST, "malformed request message")); flowCreditToRequestor(receiver, replyTo); TracingHelper.logError(currentSpan, "malformed request message"); currentSpan.finish(); return; } ProtonHelper.accepted(delivery, true); currentSpan.log("request message accepted"); final Future<ProtonSender> sender = getSenderForConnection(con, replyTo); final Future<EventBusMessage> request = createEventBusRequestMessage(requestMessage, targetAddress, clientPrincipal); CompositeFuture.all(request, sender) .compose(ok -> isAuthorized(clientPrincipal, targetAddress, requestMessage)) .map(authorized -> { logger.debug("client [{}] is {}authorized to {}:{}", clientPrincipal.getName(), authorized ? "" : "not ", targetAddress, requestMessage.getSubject()); if (authorized) { return authorized; } else { throw new ClientErrorException(HttpURLConnection.HTTP_FORBIDDEN, "not authorized to invoke operation"); } }) .compose(authorized -> { final Promise<io.vertx.core.eventbus.Message<Object>> reply = Promise.promise(); final DeliveryOptions options = createEventBusMessageDeliveryOptions(config.getSendTimeOut(), currentSpan.context()); vertx.eventBus().request( getEventBusServiceAddress(), request.result().toJson(), options, reply); return reply.future(); }) .map(reply -> extractResponse(reply)) .compose(eventBusMessage -> filterResponse(clientPrincipal, eventBusMessage)) .map(filteredResponse -> { Tags.HTTP_STATUS.set(currentSpan, filteredResponse.getStatus()); return getAmqpReply(filteredResponse); }) .otherwise(t -> { logger.debug("error processing request [resource: {}, op: {}]: {}", targetAddress, requestMessage.getSubject(), t.getMessage()); currentSpan.log("error processing request"); TracingHelper.logError(currentSpan, t); final ServiceInvocationException ex = getServiceInvocationException(t); Tags.HTTP_STATUS.set(currentSpan, ex.getErrorCode()); return RequestResponseApiConstants.getErrorMessage(ex.getErrorCode(), ex.getMessage(), requestMessage); }) .map(amqpMessage -> { if (sender.result().isOpen()) { final ProtonDelivery responseDelivery = sender.result().send(amqpMessage); logger.debug("sent response message to client [correlation-id: {}, content-type: {}]", amqpMessage.getCorrelationId(), amqpMessage.getContentType()); currentSpan.log("sent response message to client"); return responseDelivery; } else { TracingHelper.logError(currentSpan, "cannot send response, reply-to link is closed"); return null; } }) .onComplete(s -> { flowCreditToRequestor(receiver, replyTo); currentSpan.finish(); }); } protected  RequestResponseEndpoint(final Vertx vertx); final AuthorizationService getAuthorizationService(); @Autowired(required = false) final void setAuthorizationService(final AuthorizationService authService); @Override final void onLinkAttach(final ProtonConnection con, final ProtonReceiver receiver, final ResourceIdentifier targetAddress); @Override final void onLinkAttach(final ProtonConnection con, final ProtonSender sender,
            final ResourceIdentifier replyToAddress); @Override void onConnectionClosed(final ProtonConnection connection);  }
@Test public void testHandleMessageRejectsMalformedMessage() { final Message msg = ProtonHelper.message(); final ProtonConnection con = mock(ProtonConnection.class); final ProtonDelivery delivery = mock(ProtonDelivery.class); final RequestResponseEndpoint<ServiceConfigProperties> endpoint = getEndpoint(false); endpoint.handleRequestMessage(con, receiver, resource, delivery, msg); final ArgumentCaptor<DeliveryState> deliveryState = ArgumentCaptor.forClass(DeliveryState.class); verify(delivery).disposition(deliveryState.capture(), eq(Boolean.TRUE)); assertThat(deliveryState.getValue()).isInstanceOf(Rejected.class); verify(receiver, never()).close(); verify(receiver).flow(1); }