RequestResponseEndpoint extends AbstractAmqpEndpoint<T> { @Override public final void onLinkAttach(final ProtonConnection con, final ProtonReceiver receiver, final ResourceIdentifier targetAddress) { if (ProtonQoS.AT_MOST_ONCE.equals(receiver.getRemoteQoS())) { logger.debug("client wants to use unsupported AT MOST ONCE delivery mode for endpoint [{}], closing link ...", getName()); receiver.setCondition(ProtonHelper.condition(AmqpError.PRECONDITION_FAILED.toString(), "endpoint requires AT_LEAST_ONCE QoS")); receiver.close(); } else { logger.debug("establishing link for receiving request messages from client [{}]", receiver.getName()); receiver.setQoS(ProtonQoS.AT_LEAST_ONCE); receiver.setAutoAccept(true); receiver.setTarget(receiver.getRemoteTarget()); receiver.setSource(receiver.getRemoteSource()); receiver.setPrefetch(0); receiver.handler((delivery, message) -> { try { handleRequestMessage(con, receiver, targetAddress, delivery, message); } catch (final Exception ex) { logger.warn("error handling message", ex); ProtonHelper.released(delivery, true); } }); HonoProtonHelper.setCloseHandler(receiver, remoteClose -> onLinkDetach(receiver)); HonoProtonHelper.setDetachHandler(receiver, remoteDetach -> onLinkDetach(receiver)); receiver.open(); logger.debug("flowing {} credits to client", config.getReceiverLinkCredit()); receiver.flow(config.getReceiverLinkCredit()); } } 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 testOnLinkAttachClosesReceiverUsingAtMostOnceQoS() { final RequestResponseEndpoint<ServiceConfigProperties> endpoint = getEndpoint(true); when(receiver.getRemoteQoS()).thenReturn(ProtonQoS.AT_MOST_ONCE); endpoint.onLinkAttach(connection, receiver, resource); verify(receiver).close(); }
@Test public void testOnLinkAttachOpensReceiver() { final RequestResponseEndpoint<ServiceConfigProperties> endpoint = getEndpoint(true); when(receiver.getRemoteQoS()).thenReturn(ProtonQoS.AT_LEAST_ONCE); endpoint.onLinkAttach(connection, receiver, resource); verify(receiver).handler(any(ProtonMessageHandler.class)); verify(receiver).open(); verify(receiver, never()).close(); }
@Test public void testOnLinkAttachClosesSenderWithoutAppropriateReplyAddress() { final RequestResponseEndpoint<ServiceConfigProperties> endpoint = getEndpoint(true); endpoint.onLinkAttach(connection, sender, resource); verify(sender).setCondition(any()); verify(sender).close(); }
@Test public void testDuplicateSubscription() { final ProtonConnection con1 = mock(ProtonConnection.class); final ProtonConnection con2 = mock(ProtonConnection.class); final ProtonSender sender1 = mock(ProtonSender.class); final ProtonSender sender2 = mock(ProtonSender.class); final RequestResponseEndpoint<ServiceConfigProperties> endpoint = getEndpoint(true); endpoint.onLinkAttach(con1, sender1, REPLY_RESOURCE); verify(sender1).open(); endpoint.onLinkAttach(con2, sender2, REPLY_RESOURCE); verify(sender2).close(); }