import org.apache.qpid.proton.amqp.transport.AmqpError; //导入依赖的package包/类
@Override
public final void onLinkAttach(final ProtonConnection con, final ProtonSender sender, final ResourceIdentifier targetResource) {

    if (ProtonQoS.AT_LEAST_ONCE.equals(sender.getRemoteQoS())) {
        HonoUser user = Constants.getClientPrincipal(con);
        sender.setQoS(ProtonQoS.AT_LEAST_ONCE).open();
        logger.debug("transferring token to client...");
        Message tokenMsg = ProtonHelper.message(user.getToken());
        MessageHelper.addProperty(tokenMsg, AuthenticationConstants.APPLICATION_PROPERTY_TYPE, AuthenticationConstants.TYPE_AMQP_JWT);
        sender.send(tokenMsg, disposition -> {
            if (disposition.remotelySettled()) {
                logger.debug("successfully transferred auth token to client");
            } else {
                logger.debug("failed to transfer auth token to client");
            }
            sender.close();
        });
    } else {
        onLinkDetach(sender, ProtonHelper.condition(AmqpError.INVALID_FIELD, "supports AT_LEAST_ONCE delivery mode only"));
    }
}
 

import org.apache.qpid.proton.amqp.transport.AmqpError; //导入依赖的package包/类
/**
 * Closes an expired client connection.
 * <p>
 * A connection is considered expired if the {@link HonoUser#isExpired()} method
 * of the user principal attached to the connection returns {@code true}.
 * 
 * @param con The client connection.
 */
protected final void closeExpiredConnection(final ProtonConnection con) {

    if (!con.isDisconnected()) {
        final HonoUser clientPrincipal = Constants.getClientPrincipal(con);
        if (clientPrincipal != null) {
            LOG.debug("client's [{}] access token has expired, closing connection", clientPrincipal.getName());
            con.disconnectHandler(null);
            con.closeHandler(null);
            con.setCondition(ProtonHelper.condition(AmqpError.UNAUTHORIZED_ACCESS, "access token expired"));
            con.close();
            con.disconnect();
            publishConnectionClosedEvent(con);
        }
    }
}
 

import org.apache.qpid.proton.amqp.transport.AmqpError; //导入依赖的package包/类
/**
 * Configure and check the receiver link of the endpoint.
 * The remote link of the receiver must not demand the AT_MOST_ONCE QoS (not supported).
 * The receiver link itself is configured with the AT_LEAST_ONCE QoS and grants the configured credits ({@link #setReceiverLinkCredit(int)})
 * with autoAcknowledge.
 * <p>
 * Handling of received messages is delegated to {@link #handleMessage(ProtonConnection, ProtonReceiver, ResourceIdentifier, ProtonDelivery, Message)}.
 *
 * @param con The AMQP connection that the link is part of.
 * @param receiver The ProtonReceiver that has already been created for this endpoint.
 * @param targetAddress The resource identifier for this endpoint (see {@link ResourceIdentifier} for details).
 */
@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(condition(AmqpError.PRECONDITION_FAILED.toString(), "endpoint requires AT_LEAST_ONCE QoS"));
        receiver.close();
    } else {

        logger.debug("establishing link for receiving messages from client [{}]", receiver.getName());
        receiver
                .setQoS(ProtonQoS.AT_LEAST_ONCE)
                .setAutoAccept(true) // settle received messages if the handler succeeds
                .setPrefetch(receiverLinkCredit)
                .handler((delivery, message) -> {
                    handleMessage(con, receiver, targetAddress, delivery, message);
                }).closeHandler(clientDetached -> onLinkDetach(receiver))
                .open();
    }
}
 

import org.apache.qpid.proton.amqp.transport.AmqpError; //导入依赖的package包/类
@Override
public void onResponse(IAmqpProtocolConverter converter, Response response) throws IOException {
    if (response.isException()) {
        sender.setSource(null);
        Throwable exception = ((ExceptionResponse) response).getException();
        Symbol condition = AmqpError.INTERNAL_ERROR;
        if (exception instanceof InvalidSelectorException) {
            condition = AmqpError.INVALID_FIELD;
        }
        sender.setCondition(new ErrorCondition(condition, exception.getMessage()));
        subscriptionsByConsumerId.remove(id);
        sender.close();
    } else {
        sessionContext.consumers.put(id, consumerContext);
        sender.open();
    }
    pumpProtonToSocket();
}
 

import org.apache.qpid.proton.amqp.transport.AmqpError; //导入依赖的package包/类
@Override
public int sendMessage(MessageReference ref, Message message, ServerConsumer consumer, int deliveryCount) {

   ProtonServerSenderContext plugSender = (ProtonServerSenderContext) consumer.getProtocolContext();

   try {
      return plugSender.deliverMessage(ref, deliveryCount, transportConnection);
   } catch (Exception e) {
      connection.lock();
      try {
         plugSender.getSender().setCondition(new ErrorCondition(AmqpError.INTERNAL_ERROR, e.getMessage()));
         connection.flush();
      } finally {
         connection.unlock();
      }
      throw new IllegalStateException("Can't deliver message " + e, e);
   }

}
 

import org.apache.qpid.proton.amqp.transport.AmqpError; //导入依赖的package包/类
public boolean validateConnection(org.apache.qpid.proton.engine.Connection connection, SASLResult saslResult) {
   remoteContainerId = connection.getRemoteContainer();
   boolean idOK = server.addClientConnection(remoteContainerId, ExtCapability.needUniqueConnection(connection));
   if (!idOK) {
      //https://issues.apache.org/jira/browse/ARTEMIS-728
      Map<Symbol, Object> connProp = new HashMap<>();
      connProp.put(AmqpSupport.CONNECTION_OPEN_FAILED, "true");
      connection.setProperties(connProp);
      connection.getCondition().setCondition(AmqpError.INVALID_FIELD);
      Map<Symbol, Symbol> info = new HashMap<>();
      info.put(AmqpSupport.INVALID_FIELD, AmqpSupport.CONTAINER_ID);
      connection.getCondition().setInfo(info);
      return false;
   }
   registeredConnectionId.set(true);
   return true;
}
 

import org.apache.qpid.proton.amqp.transport.AmqpError; //导入依赖的package包/类
@Test
public void testCreateRedirectionExceptionWithNoNetworkHost() throws URISyntaxException {
    AmqpProvider mockProvider = Mockito.mock(AmqpProvider.class);
    Mockito.when(mockProvider.getRemoteURI()).thenReturn(new URI("amqp://localhost:5672"));

    ErrorCondition condition = new ErrorCondition();

    Map<Symbol, Object> info = new HashMap<>();
    info.put(AmqpSupport.PORT, "5672");
    info.put(AmqpSupport.OPEN_HOSTNAME, "localhost");
    info.put(AmqpSupport.SCHEME, "amqp");
    info.put(AmqpSupport.PATH, "websocket");

    condition.setInfo(info);

    Symbol error = AmqpError.INTERNAL_ERROR;
    String message = "Failed to connect";

    Exception result = AmqpSupport.createRedirectException(mockProvider, error, message, condition);

    assertNotNull(result);
    assertFalse(result instanceof ProviderRedirectedException);
    assertTrue(result instanceof IOException);
}
 

import org.apache.qpid.proton.amqp.transport.AmqpError; //导入依赖的package包/类
@Test
public void testCreateRedirectionExceptionWithEmptyNetworkHost() throws URISyntaxException {
    AmqpProvider mockProvider = Mockito.mock(AmqpProvider.class);
    Mockito.when(mockProvider.getRemoteURI()).thenReturn(new URI("amqp://localhost:5672"));

    ErrorCondition condition = new ErrorCondition();

    Map<Symbol, Object> info = new HashMap<>();
    info.put(AmqpSupport.PORT, "5672");
    info.put(AmqpSupport.NETWORK_HOST, "");
    info.put(AmqpSupport.OPEN_HOSTNAME, "localhost");
    info.put(AmqpSupport.SCHEME, "amqp");
    info.put(AmqpSupport.PATH, "websocket");

    condition.setInfo(info);

    Symbol error = AmqpError.INTERNAL_ERROR;
    String message = "Failed to connect";

    Exception result = AmqpSupport.createRedirectException(mockProvider, error, message, condition);

    assertNotNull(result);
    assertFalse(result instanceof ProviderRedirectedException);
    assertTrue(result instanceof IOException);
}
 

import org.apache.qpid.proton.amqp.transport.AmqpError; //导入依赖的package包/类
@Test
public void testCreateRedirectionExceptionWithInvalidPort() throws URISyntaxException {
    AmqpProvider mockProvider = Mockito.mock(AmqpProvider.class);
    Mockito.when(mockProvider.getRemoteURI()).thenReturn(new URI("amqp://localhost:5672"));

    ErrorCondition condition = new ErrorCondition();

    Map<Symbol, Object> info = new HashMap<>();
    info.put(AmqpSupport.PORT, "L5672");
    info.put(AmqpSupport.OPEN_HOSTNAME, "localhost");
    info.put(AmqpSupport.NETWORK_HOST, "localhost");
    info.put(AmqpSupport.SCHEME, "amqp");
    info.put(AmqpSupport.PATH, "websocket");

    condition.setInfo(info);

    Symbol error = AmqpError.INTERNAL_ERROR;
    String message = "Failed to connect";

    Exception result = AmqpSupport.createRedirectException(mockProvider, error, message, condition);

    assertNotNull(result);
    assertFalse(result instanceof ProviderRedirectedException);
    assertTrue(result instanceof IOException);
}
 

import org.apache.qpid.proton.amqp.transport.AmqpError; //导入依赖的package包/类
/**
 * Handles a request from a client to establish a link for receiving messages from this server.
 *
 * @param con the connection to the client.
 * @param sender the sender created for the link.
 */
@Override
protected void handleSenderOpen(final ProtonConnection con, final ProtonSender sender) {

    final Source remoteSource = sender.getRemoteSource();
    LOG.debug("client [{}] wants to open a link for receiving messages [address: {}]",
            con.getRemoteContainer(), remoteSource);
    try {
        final ResourceIdentifier targetResource = getResourceIdentifier(remoteSource.getAddress());
        final AmqpEndpoint endpoint = getEndpoint(targetResource);

        if (endpoint == null) {
            LOG.debug("no endpoint registered for node [{}]", targetResource);
            con.setCondition(ProtonHelper.condition(AmqpError.NOT_FOUND, "no such node")).close();
        } else {
            HonoUser user = Constants.getClientPrincipal(con);
            if (Constants.SUBJECT_ANONYMOUS.equals(user.getName())) {
                con.setCondition(ProtonHelper.condition(AmqpError.UNAUTHORIZED_ACCESS, "client must authenticate using SASL")).close();
            } else {
                Constants.copyProperties(con, sender);
                sender.setSource(sender.getRemoteSource());
                endpoint.onLinkAttach(con, sender, targetResource);
                vertx.setTimer(5000, closeCon -> {
                    if (!con.isDisconnected()) {
                        LOG.debug("connection with client [{}] timed out after 5 seconds, closing connection", con.getRemoteContainer());
                        con.setCondition(ProtonHelper.condition("hono: inactivity", "client must retrieve token within 5 secs after opening connection")).close();
                    }
                });
            }
        }
    } catch (final IllegalArgumentException e) {
        LOG.debug("client has provided invalid resource identifier as source address", e);
        con.setCondition(ProtonHelper.condition(AmqpError.INVALID_FIELD, "malformed source address")).close();
    }
}
 

import org.apache.qpid.proton.amqp.transport.AmqpError; //导入依赖的package包/类
@Override
public final void onLinkAttach(final ProtonConnection con, final ProtonReceiver receiver, final ResourceIdentifier targetAddress) {

    if (!Arrays.stream(getEndpointQos()).anyMatch(qos -> qos.equals(receiver.getRemoteQoS()))) {
        logger.debug("client [{}] wants to use unsupported delivery mode {} for endpoint [name: {}, QoS: {}], closing link", 
                con.getRemoteContainer(), receiver.getRemoteQoS(), getName(), getEndpointQos());
        receiver.setCondition(ErrorConditions.ERROR_UNSUPPORTED_DELIVERY_MODE);
        receiver.close();
    } else {
        receiver.setQoS(receiver.getRemoteQoS());
        receiver.setTarget(receiver.getRemoteTarget());
        final String linkId = UUID.randomUUID().toString();
        final UpstreamReceiver link = UpstreamReceiver.newUpstreamReceiver(linkId, receiver);

        downstreamAdapter.onClientAttach(link, s -> {
            if (s.succeeded()) {
                receiver.closeHandler(clientDetached -> {
                    // client has closed link -> inform TelemetryAdapter about client detach
                    onLinkDetach(link);
                    downstreamAdapter.onClientDetach(link);
                    metrics.decrementUpstreamLinks(targetAddress.toString());
                }).handler((delivery, message) -> {
                    if (passesFormalVerification(targetAddress, message)) {
                        forwardMessage(link, delivery, message);
                    } else {
                        rejectMessage(delivery, ProtonHelper.condition(AmqpError.DECODE_ERROR, "malformed message"), link);
                    }
                }).open();
                logger.debug("establishing link with client [{}]", con.getRemoteContainer());
                metrics.incrementUpstreamLinks(targetAddress.toString());
            } else {
                // we cannot connect to downstream container, reject client
                link.close(condition(AmqpError.PRECONDITION_FAILED, "no consumer available for target"));
            }
        });
    }
}
 

import org.apache.qpid.proton.amqp.transport.AmqpError; //导入依赖的package包/类
final void forwardMessage(final UpstreamReceiver link, final ProtonDelivery delivery, final Message msg) {

        final ResourceIdentifier messageAddress = ResourceIdentifier.fromString(getAnnotation(msg, MessageHelper.APP_PROPERTY_RESOURCE, String.class));
        final String token = MessageHelper.getAndRemoveRegistrationAssertion(msg);

        if (assertRegistration(token, messageAddress)) {
            downstreamAdapter.processMessage(link, delivery, msg);
        } else {
            logger.debug("failed to validate device registration status");
            rejectMessage(delivery, ProtonHelper.condition(AmqpError.PRECONDITION_FAILED, "device non-existent/disabled"), link);
        }
    }
 

import org.apache.qpid.proton.amqp.transport.AmqpError; //导入依赖的package包/类
/**
 * Verifies that the endpoint rejects messages that do not pass formal verification.
 */
@SuppressWarnings("unchecked")
@Test
public void testMessageHandlerRejectsMalformedMessage() {

    // GIVEN an endpoint with an attached client
    final ProtonConnection connection = mock(ProtonConnection.class);
    when(connection.getRemoteContainer()).thenReturn("test-client");
    final ProtonReceiver receiver = mock(ProtonReceiver.class);
    final ResourceIdentifier targetAddress = ResourceIdentifier.fromString("telemetry/tenant");
    ArgumentCaptor<ProtonMessageHandler> messageHandler = ArgumentCaptor.forClass(ProtonMessageHandler.class);
    when(receiver.handler(messageHandler.capture())).thenReturn(receiver);
    when(receiver.closeHandler(any(Handler.class))).thenReturn(receiver);
    when(receiver.getRemoteQoS()).thenReturn(ProtonQoS.AT_MOST_ONCE);
    final DownstreamAdapter adapter = mock(DownstreamAdapter.class);
    doAnswer(invocation -> {
        invocation.getArgumentAt(1, Handler.class).handle(Future.succeededFuture(null));
        return null;
    }).when(adapter).onClientAttach(any(UpstreamReceiver.class), any(Handler.class));

    MessageForwardingEndpoint<ServiceConfigProperties> endpoint = getEndpoint(false);
    endpoint.setDownstreamAdapter(adapter);
    endpoint.onLinkAttach(connection, receiver, targetAddress);

    // WHEN a client sends a malformed message
    Message message = ProtonHelper.message("malformed");
    ProtonDelivery upstreamDelivery = mock(ProtonDelivery.class);
    messageHandler.getValue().handle(upstreamDelivery, message);

    // THEN a the endpoint rejects the message
    ArgumentCaptor<Rejected> deliveryState = ArgumentCaptor.forClass(Rejected.class);
    verify(upstreamDelivery).disposition(deliveryState.capture(), eq(Boolean.TRUE));
    assertThat(deliveryState.getValue().getError().getCondition(), is(AmqpError.DECODE_ERROR));
    // but does not close the link
    verify(receiver, never()).close();
}
 

import org.apache.qpid.proton.amqp.transport.AmqpError; //导入依赖的package包/类
/**
 * Verifies that the endpoint does not open a link with a client if the
 * downstream messaging network is not available.
 */
@SuppressWarnings("unchecked")
@Test
public void testOnLinkAttachClosesLinkIfDownstreamIsNotAvailable() {

    // GIVEN an endpoint without a connection to the downstream messaging network
    final ProtonConnection connection = mock(ProtonConnection.class);
    final ProtonReceiver receiver = mock(ProtonReceiver.class);
    final ResourceIdentifier targetAddress = ResourceIdentifier.fromString("telemetry/tenant");
    when(receiver.getRemoteQoS()).thenReturn(ProtonQoS.AT_MOST_ONCE);
    final DownstreamAdapter adapter = mock(DownstreamAdapter.class);
    doAnswer(invocation -> {
        invocation.getArgumentAt(1, Handler.class).handle(Future.failedFuture("downstream not available"));
        return null;
    }).when(adapter).onClientAttach(any(UpstreamReceiver.class), any(Handler.class));

    MessageForwardingEndpoint<ServiceConfigProperties> endpoint = getEndpoint(false);
    endpoint.setDownstreamAdapter(adapter);

    // WHEN a client tries to attach
    endpoint.onLinkAttach(connection, receiver, targetAddress);

    // THEN the endpoint closes the link
    ArgumentCaptor<ErrorCondition> errorCondition = ArgumentCaptor.forClass(ErrorCondition.class);
    verify(receiver).setCondition(errorCondition.capture());
    assertThat(errorCondition.getValue().getCondition(), is(AmqpError.PRECONDITION_FAILED));
    verify(receiver).close();
}
 

import org.apache.qpid.proton.amqp.transport.AmqpError; //导入依赖的package包/类
/**
 * Closes a link for an unknown target address.
 * <p>
 * The link is closed with AMQP error code <em>amqp:not-found</em>.
 * 
 * @param con The connection that the link belongs to.
 * @param link The link.
 * @param address The unknown target address.
 */
protected final void handleUnknownEndpoint(final ProtonConnection con, final ProtonLink<?> link, final ResourceIdentifier address) {
    LOG.info("client [container: {}] wants to establish link for unknown endpoint [address: {}]",
            con.getRemoteContainer(), address);
    link.setCondition(
            ProtonHelper.condition(
                    AmqpError.NOT_FOUND,
                    String.format("no endpoint registered for address %s", address)));
    link.close();
}
 

import org.apache.qpid.proton.amqp.transport.AmqpError; //导入依赖的package包/类
/**
 * Handles a request from a client to establish a link for sending messages to this server.
 * The already established connection must have an authenticated user as principal for doing the authorization check.
 *
 * @param con the connection to the client.
 * @param receiver the receiver created for the link.
 */
protected void handleReceiverOpen(final ProtonConnection con, final ProtonReceiver receiver) {
    if (receiver.getRemoteTarget().getAddress() == null) {
        LOG.debug("client [container: {}] wants to open an anonymous link for sending messages to arbitrary addresses, closing link ...",
                con.getRemoteContainer());
        receiver.setCondition(ProtonHelper.condition(AmqpError.NOT_ALLOWED, "anonymous relay not supported"));
        receiver.close();
    } else {
        LOG.debug("client [container: {}] wants to open a link [address: {}] for sending messages",
                con.getRemoteContainer(), receiver.getRemoteTarget());
        try {
            final ResourceIdentifier targetResource = getResourceIdentifier(receiver.getRemoteTarget().getAddress());
            final AmqpEndpoint endpoint = getEndpoint(targetResource);
            if (endpoint == null) {
                handleUnknownEndpoint(con, receiver, targetResource);
            } else {
                final HonoUser user = Constants.getClientPrincipal(con);
                getAuthorizationService().isAuthorized(user, targetResource, Activity.WRITE).setHandler(authAttempt -> {
                    if (authAttempt.succeeded() && authAttempt.result()) {
                        Constants.copyProperties(con, receiver);
                        receiver.setTarget(receiver.getRemoteTarget());
                        endpoint.onLinkAttach(con, receiver, targetResource);
                    } else {
                        LOG.debug("subject [{}] is not authorized to WRITE to [{}]", user.getName(), targetResource);
                        receiver.setCondition(ProtonHelper.condition(AmqpError.UNAUTHORIZED_ACCESS.toString(), "unauthorized"));
                        receiver.close();
                    }
                });
            }
        } catch (final IllegalArgumentException e) {
            LOG.debug("client has provided invalid resource identifier as target address", e);
            receiver.setCondition(ProtonHelper.condition(AmqpError.NOT_FOUND, "no such address"));
            receiver.close();
        }
    }
}
 

import org.apache.qpid.proton.amqp.transport.AmqpError; //导入依赖的package包/类
/**
 * Handles a request from a client to establish a link for receiving messages from this server.
 *
 * @param con the connection to the client.
 * @param sender the sender created for the link.
 */
protected void handleSenderOpen(final ProtonConnection con, final ProtonSender sender) {
    final Source remoteSource = sender.getRemoteSource();
    LOG.debug("client [container: {}] wants to open a link [address: {}] for receiving messages",
            con.getRemoteContainer(), remoteSource);
    try {
        final ResourceIdentifier targetResource = getResourceIdentifier(remoteSource.getAddress());
        final AmqpEndpoint endpoint = getEndpoint(targetResource);
        if (endpoint == null) {
            handleUnknownEndpoint(con, sender, targetResource);
        } else {
            final HonoUser user = Constants.getClientPrincipal(con);
            getAuthorizationService().isAuthorized(user, targetResource, Activity.READ).setHandler(authAttempt -> {
                if (authAttempt.succeeded() && authAttempt.result()) {
                    Constants.copyProperties(con, sender);
                    sender.setSource(sender.getRemoteSource());
                    endpoint.onLinkAttach(con, sender, targetResource);
                } else {
                    LOG.debug("subject [{}] is not authorized to READ from [{}]", user.getName(), targetResource);
                    sender.setCondition(ProtonHelper.condition(AmqpError.UNAUTHORIZED_ACCESS.toString(), "unauthorized"));
                    sender.close();
                }
            });
        }
    } catch (final IllegalArgumentException e) {
        LOG.debug("client has provided invalid resource identifier as target address", e);
        sender.setCondition(ProtonHelper.condition(AmqpError.NOT_FOUND, "no such address"));
        sender.close();
    }
}
 

import org.apache.qpid.proton.amqp.transport.AmqpError; //导入依赖的package包/类
private void receiverHandler(ProtonReceiver receiver) {

        LOG.info("Attaching link request");

        // the LWT service supports only the control address
        if (!receiver.getRemoteTarget().getAddress().equals(LWT_SERVICE_ENDPOINT)) {

            ErrorCondition errorCondition =
                    new ErrorCondition(AmqpError.NOT_FOUND, "The provided address isn't supported");

            receiver.setCondition(errorCondition)
                    .close();
        } else {

            receiver.setTarget(receiver.getRemoteTarget())
                    .setQoS(ProtonQoS.AT_LEAST_ONCE)
                    .handler((delivery, message) -> {
                        this.messageHandler(receiver, delivery, message);
                    })
                    .closeHandler(ar -> {
                        this.closeHandler(receiver, ar);
                    })
                    .detachHandler(ar -> {
                        this.closeHandler(receiver, ar);
                    })
                    .setPrefetch(0)
                    .open();

            receiver.flow(AMQP_WILL_CREDITS);
        }
    }
 

import org.apache.qpid.proton.amqp.transport.AmqpError; //导入依赖的package包/类
@Test
public void attachingAddressNotSupported(TestContext context) throws InterruptedException {


    // deploy the MQTT LWT service
    super.deploy(context);
    // trying to attach a not supported address by MQTT LWT service
    // NOTE : $lwt.wrong passes through the router (link route "prefix" is $lwt)
    //        but it should be refused by the MQTT LWT service
    this.attaching(context, LWT_SERVICE_ENDPOINT + ".wrong", AmqpError.NOT_FOUND);
}
 

import org.apache.qpid.proton.amqp.transport.AmqpError; //导入依赖的package包/类
private void receiverHandler(ProtonReceiver receiver) {

        // Last Will and Testament Service supports only the control address
        if (!receiver.getRemoteTarget().getAddress().equals(LWT_SERVICE_ENDPOINT)) {

            ErrorCondition error = new ErrorCondition(AmqpError.NOT_FOUND, "The endpoint provided is not supported");
            receiver
                    .setCondition(error)
                    .close();
        } else {

            // TODO: tracking the AMQP sender ??

            receiver
                    .setTarget(receiver.getRemoteTarget())
                    .setQoS(ProtonQoS.AT_LEAST_ONCE)
                    .handler((delivery, message) -> {

                        this.messageHandler(receiver, delivery, message);
                    })
                    .closeHandler(ar -> {

                        this.closeHandler(receiver, ar);
                    })
                    .open();
        }
    }