import io.vertx.proton.ProtonDelivery; //导入依赖的package包/类
private void receiverHandler(ProtonReceiver receiver, ProtonDelivery delivery, Message message) {

        Section section = message.getBody();

        byte[] data = null;
        if (section instanceof AmqpValue) {
            data = ((String) ((AmqpValue)section).getValue()).getBytes();
        } else if (section instanceof Data) {
            data = ((Data)message.getBody()).getValue().getArray();
        } else {
            log.error("Discarded message : body type not supported");
        }

        MessageDelivery messageDelivery =
                new MessageDelivery(receiver.getSource().getAddress(), data);

        delivery.disposition(Accepted.getInstance(), true);

        this.receivedHandler.handle(messageDelivery);
    }
 

import io.vertx.proton.ProtonDelivery; //导入依赖的package包/类
@Override
public void processMessage(final UpstreamReceiver client, final ProtonDelivery delivery, final Message data) {

    LinkStatus status = statusMap.get(client.getLinkId());
    if (status == null) {
        LOG.debug("creating new link status object [{}]", client.getLinkId());
        status = new LinkStatus(client);
        statusMap.put(client.getLinkId(), status);
    }
    LOG.debug("processing telemetry data [id: {}, to: {}, content-type: {}]", data.getMessageId(), data.getAddress(),
            data.getContentType());
    if (messageConsumer != null) {
        messageConsumer.accept(data);
    }
    ProtonHelper.accepted(delivery, true);
    status.onMsgReceived();
}
 

import io.vertx.proton.ProtonDelivery; //导入依赖的package包/类
private void givenADownstreamAdapter(final SenderFactory senderFactory) {

        adapter = new ForwardingDownstreamAdapter(vertx, senderFactory) {

            @Override
            protected ProtonQoS getDownstreamQos() {
                return ProtonQoS.AT_MOST_ONCE;
            }

            @Override
            protected void forwardMessage(final ProtonSender sender, final Message msg, final ProtonDelivery delivery) {
                // nothing to do
            }
        };
        adapter.setMetrics(mock(MessagingMetrics.class));
    }
 

import io.vertx.proton.ProtonDelivery; //导入依赖的package包/类
/**
 * Verifies that a message containing a matching registration assertion is
 * forwarded to the downstream adapter.
 */
@Test
public void testForwardMessageAcceptsCorrectRegistrationAssertion() {

    final String validToken = getToken(SECRET, "tenant", "4711");
    UpstreamReceiver client = mock(UpstreamReceiver.class);
    ProtonDelivery delivery = mock(ProtonDelivery.class);
    DownstreamAdapter adapter = mock(DownstreamAdapter.class);
    when(tokenValidator.isValid(validToken, "tenant", "4711")).thenReturn(Boolean.TRUE);
    MessageForwardingEndpoint<ServiceConfigProperties> endpoint = getEndpoint();
    endpoint.setRegistrationAssertionValidator(tokenValidator);
    endpoint.setDownstreamAdapter(adapter);

    // WHEN processing a message bearing a valid registration assertion
    Message msg = ProtonHelper.message();
    MessageHelper.addRegistrationAssertion(msg, validToken);
    MessageHelper.addAnnotation(msg, MessageHelper.APP_PROPERTY_RESOURCE, "telemetry/tenant/4711");
    endpoint.forwardMessage(client, delivery, msg);

    // THEN the message is sent downstream
    verify(adapter).processMessage(client, delivery, msg);
    verify(client, times(0)).close(any(ErrorCondition.class));
    // and the assertion has been removed from the message
    assertThat("downstream message should not contain registration assertion",
            MessageHelper.getRegistrationAssertion(msg), is(nullValue()));
}
 

import io.vertx.proton.ProtonDelivery; //导入依赖的package包/类
/**
 * Verifies that a message containing a registration assertion for a tenant
 * other than the one from the message's target address is rejected.
 */
@Test
public void testProcessMessageRejectsRegistrationAssertionForWrongTenant() {

    final String invalidToken = getToken(SECRET, "wrong-tenant", "4711");
    UpstreamReceiver client = mock(UpstreamReceiver.class);
    ProtonDelivery delivery = mock(ProtonDelivery.class);
    when(tokenValidator.isValid(invalidToken, "tenant", "4711")).thenReturn(Boolean.FALSE);
    MessageForwardingEndpoint<ServiceConfigProperties> endpoint = getEndpoint();
    endpoint.setRegistrationAssertionValidator(tokenValidator);

    Message msg = ProtonHelper.message();
    MessageHelper.addRegistrationAssertion(msg, invalidToken);
    MessageHelper.addAnnotation(msg, MessageHelper.APP_PROPERTY_RESOURCE, "telemetry/tenant/4711");
    endpoint.forwardMessage(client, delivery, msg);

    verify(delivery).disposition(any(Rejected.class), anyBoolean());
    verify(client, never()).close(any(ErrorCondition.class));
}
 

import io.vertx.proton.ProtonDelivery; //导入依赖的package包/类
@Override
public final Future<ProtonDelivery> send(final String deviceId, final Map<String, ?> properties, final byte[] payload, final String contentType,
                          final String registrationAssertion) {
    Objects.requireNonNull(deviceId);
    Objects.requireNonNull(payload);
    Objects.requireNonNull(contentType);
    Objects.requireNonNull(registrationAssertion);

    final Message msg = ProtonHelper.message();
    msg.setAddress(getTo(deviceId));
    msg.setBody(new Data(new Binary(payload)));
    setApplicationProperties(msg, properties);
    addProperties(msg, deviceId, contentType, registrationAssertion);
    addEndpointSpecificProperties(msg, deviceId);
    return send(msg);
}
 

import io.vertx.proton.ProtonDelivery; //导入依赖的package包/类
/**
 * Handles a response received from the peer.
 * <p>
 * In particular, this method tries to correlate the message with a previous request
 * using the message's <em>correlation-id</em> and, if successful, the delivery is <em>accepted</em>
 * and the message is passed to the handler registered with the original request.
 * <p>
 * If the response cannot be correlated to a request, e.g. because the request has timed
 * out, then the delivery is <em>released</em> and the message is silently discarded.
 * 
 * @param delivery The handle for accessing the message's disposition.
 * @param message The response message.
 */
protected final void handleResponse(final ProtonDelivery delivery, final Message message) {

    final Handler<AsyncResult<R>> handler = replyMap.remove(message.getCorrelationId());
    if (handler != null) {
        R response = getRequestResponseResult(message);
        LOG.debug("received response [reply-to: {}, subject: {}, correlation ID: {}, status: {}]",
                replyToAddress, message.getSubject(), message.getCorrelationId(), response.getStatus());
        handler.handle(Future.succeededFuture(response));
        ProtonHelper.accepted(delivery, true);
    } else {
        LOG.debug("discarding unexpected response [reply-to: {}, correlation ID: {}]",
                replyToAddress, message.getCorrelationId());
        ProtonHelper.released(delivery, true);
    }
}
 

import io.vertx.proton.ProtonDelivery; //导入依赖的package包/类
@Override
public HonoClient createEventConsumer(
        final String tenantId,
        final BiConsumer<ProtonDelivery, Message> eventConsumer,
        final Handler<AsyncResult<MessageConsumer>> creationHandler) {

    // register a handler to be notified if the underlying connection to the server fails
    // so that we can fail the result handler passed in
    final Handler<Void> connectionFailureHandler = connectionLost -> {
        creationHandler.handle(Future.failedFuture(
                new ServerErrorException(HttpURLConnection.HTTP_UNAVAILABLE, "connection to server lost")));
    };
    creationRequests.add(connectionFailureHandler);

    Future<MessageConsumer> consumerTracker = Future.future();
    consumerTracker.setHandler(attempt -> {
        creationRequests.remove(connectionFailureHandler);
        creationHandler.handle(attempt);
    });
    checkConnection().compose(
            connected -> EventConsumerImpl.create(context, clientConfigProperties, connection, tenantId,
                    connectionFactory.getPathSeparator(), eventConsumer, consumerTracker.completer()),
            consumerTracker);
    return this;
}
 

import io.vertx.proton.ProtonDelivery; //导入依赖的package包/类
/**
 * Creates a new event consumer for a tenant.
 * 
 * @param context The vert.x context to run all interactions with the server on.
 * @param clientConfig The configuration properties to use.
 * @param con The AMQP connection to the server.
 * @param tenantId The tenant to consumer events for.
 * @param pathSeparator The address path separator character used by the server.
 * @param eventConsumer The consumer to invoke with each event received.
 * @param creationHandler The handler to invoke with the outcome of the creation attempt.
 * @throws NullPointerException if any of the parameters is {@code null}.
 */
public static void create(
        final Context context,
        final ClientConfigProperties clientConfig,
        final ProtonConnection con,
        final String tenantId,
        final String pathSeparator,
        final BiConsumer<ProtonDelivery, Message> eventConsumer,
        final Handler<AsyncResult<MessageConsumer>> creationHandler) {

    Objects.requireNonNull(context);
    Objects.requireNonNull(clientConfig);
    Objects.requireNonNull(con);
    Objects.requireNonNull(tenantId);
    Objects.requireNonNull(pathSeparator);
    Objects.requireNonNull(eventConsumer);
    Objects.requireNonNull(creationHandler);
    createConsumer(context, clientConfig, con, tenantId, pathSeparator, EVENT_ADDRESS_TEMPLATE, ProtonQoS.AT_LEAST_ONCE, eventConsumer).setHandler(created -> {
        if (created.succeeded()) {
            creationHandler.handle(Future.succeededFuture(
                    new EventConsumerImpl(context, clientConfig, created.result())));
        } else {
            creationHandler.handle(Future.failedFuture(created.cause()));
        }
    });
}
 

import io.vertx.proton.ProtonDelivery; //导入依赖的package包/类
/**
 * Verifies that the client passes a response message to the handler registered for the request that
 * the response correlates with.
 * 
 * @param ctx The vert.x test context.
 */
@SuppressWarnings("unchecked")
@Test
public void testHandleResponseInvokesHandlerForMatchingCorrelationId(final TestContext ctx) {

    // GIVEN a request message that has been sent to a peer
    final Async responseReceived = ctx.async();
    client.createAndSendRequest("request", null, null, ctx.asyncAssertSuccess(s -> {
        ctx.assertEquals(200, s.getStatus());
        ctx.assertEquals("payload", s.getPayload());
        responseReceived.complete();
    }));

    // WHEN a response is received for the request
    final Message response = ProtonHelper.message("payload");
    response.setCorrelationId(MESSAGE_ID);
    MessageHelper.addProperty(response, MessageHelper.APP_PROPERTY_STATUS, 200);
    final ProtonDelivery delivery = mock(ProtonDelivery.class);
    client.handleResponse(delivery, response);

    // THEN the response is passed to the handler registered with the request
    responseReceived.await(1000);
    verify(vertx, never()).setTimer(anyLong(), any(Handler.class));
}
 

import io.vertx.proton.ProtonDelivery; //导入依赖的package包/类
/**
 * Verifies that the sender fails if no credit is available.
 * 
 * @param ctx The vert.x test context.
 */
@SuppressWarnings("unchecked")
@Test
public void testSendMessageFailsOnLackOfCredit(final TestContext ctx) {

    // GIVEN a sender that has no credit
    when(sender.sendQueueFull()).thenReturn(Boolean.TRUE);
    MessageSender messageSender = new EventSenderImpl(config, sender, "tenant", "telemetry/tenant", context);

    // WHEN trying to send a message
    final Future<ProtonDelivery> result = messageSender.send("device", "some payload", "application/text", "token");

    // THEN the message is not sent
    assertFalse(result.succeeded());
    verify(sender, never()).send(any(Message.class), any(Handler.class));
}
 

import io.vertx.proton.ProtonDelivery; //导入依赖的package包/类
/**
 * Verifies that the sender fails if no credit is available.
 * 
 * @param ctx The vert.x test context.
 */
@SuppressWarnings("unchecked")
@Test
public void testSendMessageFailsOnLackOfCredit(final TestContext ctx) {

    // GIVEN a sender that has no credit
    when(sender.sendQueueFull()).thenReturn(Boolean.TRUE);
    MessageSender messageSender = new TelemetrySenderImpl(config, sender, "tenant", "telemetry/tenant", context);

    // WHEN trying to send a message
    final Future<ProtonDelivery> result = messageSender.send("device", "some payload", "application/text", "token");

    // THEN the message is not sent
    assertFalse(result.succeeded());
    verify(sender, never()).send(any(Message.class), any(Handler.class));
}
 

import io.vertx.proton.ProtonDelivery; //导入依赖的package包/类
/**
 * Verifies that the adapter does not wait for a telemetry message being settled and accepted
 * by a downstream peer.
 * 
 * @param ctx The vert.x test context.
 */
@Test
public void testOnUnauthenticatedMessageDoesNotWaitForAcceptedOutcome(final TestContext ctx) {

    // GIVEN an adapter with a downstream telemetry consumer
    final Future<ProtonDelivery> outcome = Future.succeededFuture(mock(ProtonDelivery.class));
    givenATelemetrySenderForOutcome(outcome);
    MqttServer server = getMqttServer(false);
    AbstractVertxBasedMqttProtocolAdapter<ProtocolAdapterProperties> adapter = getAdapter(server);

    // WHEN a device publishes a telemetry message
    final Buffer payload = Buffer.buffer("some payload");
    final MqttEndpoint endpoint = mock(MqttEndpoint.class);
    final MqttPublishMessage messageFromDevice = mock(MqttPublishMessage.class);
    when(messageFromDevice.topicName()).thenReturn("telemetry/my-tenant/4712");
    when(messageFromDevice.qosLevel()).thenReturn(MqttQoS.AT_MOST_ONCE);
    when(messageFromDevice.payload()).thenReturn(payload);
    adapter.onUnauthenticatedMessage(endpoint, messageFromDevice).setHandler(ctx.asyncAssertSuccess());
}
 

import io.vertx.proton.ProtonDelivery; //导入依赖的package包/类
/**
 * Verifies that the adapter waits for an event being settled and accepted
 * by a downstream peer before responding with a 202 status to the device.
 */
@Test
public void testUploadEventWaitsForAcceptedOutcome() {

    // GIVEN an adapter with a downstream event consumer attached
    final Future<ProtonDelivery> outcome = Future.future();
    givenAnEventSenderForOutcome(outcome);

    HttpServer server = getHttpServer(false);
    AbstractVertxBasedHttpProtocolAdapter<HttpProtocolAdapterProperties> adapter = getAdapter(server, null);

    // WHEN a device publishes an event
    final Buffer payload = Buffer.buffer("some payload");
    final HttpServerResponse response = mock(HttpServerResponse.class);
    final RoutingContext ctx = newRoutingContext(payload, response);

    adapter.uploadEventMessage(ctx, "tenant", "device", payload, "application/text");

    // THEN the device does not get a response
    verify(response, never()).end();

    // until the event has been accepted
    outcome.complete(mock(ProtonDelivery.class));
    verify(response).setStatusCode(202);
    verify(response).end();
}
 

import io.vertx.proton.ProtonDelivery; //导入依赖的package包/类
/**
 * Verifies that the adapter fails the upload of an event with a 400
 * result if it is rejected by the downstream peer.
 */
@Test
public void testUploadEventFailsForRejectedOutcome() {

    // GIVEN an adapter with a downstream event consumer attached
    final Future<ProtonDelivery> outcome = Future.future();
    givenAnEventSenderForOutcome(outcome);

    HttpServer server = getHttpServer(false);
    AbstractVertxBasedHttpProtocolAdapter<HttpProtocolAdapterProperties> adapter = getAdapter(server, null);

    // WHEN a device publishes an event that is not accepted by the peer
    final Buffer payload = Buffer.buffer("some payload");
    final RoutingContext ctx = newRoutingContext(payload);

    adapter.uploadEventMessage(ctx, "tenant", "device", payload, "application/text");
    outcome.fail(new ClientErrorException(HttpURLConnection.HTTP_BAD_REQUEST, "malformed message"));

    // THEN the device gets a 400
    verify(ctx).fail(HttpURLConnection.HTTP_BAD_REQUEST);
}
 

import io.vertx.proton.ProtonDelivery; //导入依赖的package包/类
/**
 * Verifies that the adapter does not wait for a telemetry message being settled and accepted
 * by a downstream peer before responding with a 202 status to the device.
 */
@Test
public void testUploadTelemetryDoesNotWaitForAcceptedOutcome() {

    // GIVEN an adapter with a downstream telemetry consumer attached
    final Future<ProtonDelivery> outcome = Future.succeededFuture(mock(ProtonDelivery.class));
    givenATelemetrySenderForOutcome(outcome);

    HttpServer server = getHttpServer(false);
    AbstractVertxBasedHttpProtocolAdapter<HttpProtocolAdapterProperties> adapter = getAdapter(server, null);

    // WHEN a device publishes a telemetry message
    final Buffer payload = Buffer.buffer("some payload");
    final HttpServerResponse response = mock(HttpServerResponse.class);
    final RoutingContext ctx = newRoutingContext(payload, response);

    adapter.uploadTelemetryMessage(ctx, "tenant", "device", payload, "application/text");

    // THEN the device receives a 202 response immediately
    verify(response).setStatusCode(202);
    verify(response).end();
}
 

import io.vertx.proton.ProtonDelivery; //导入依赖的package包/类
/**
 * Verifies that the endpoint rejects malformed request messages.
 */
@Test
public void testHandleMessageRejectsMalformedMessage() {

    Message msg = ProtonHelper.message();
    ProtonConnection con = mock(ProtonConnection.class);
    ProtonDelivery delivery = mock(ProtonDelivery.class);
    RequestResponseEndpoint<ServiceConfigProperties> endpoint = getEndpoint(false);

    // WHEN a malformed message is received
    endpoint.handleMessage(con, receiver, resource, delivery, msg);

    // THEN the link is closed and the message is rejected
    ArgumentCaptor<DeliveryState> deliveryState = ArgumentCaptor.forClass(DeliveryState.class);
    verify(delivery).disposition(deliveryState.capture(), booleanThat(is(Boolean.TRUE)));
    assertThat(deliveryState.getValue(), instanceOf(Rejected.class));
    verify(receiver, never()).close();
}
 

import io.vertx.proton.ProtonDelivery; //导入依赖的package包/类
/**
 * Verifies that the endpoint rejects request messages for operations the client
 * is not authorized to invoke.
 */
@Test
public void testHandleMessageRejectsUnauthorizedRequests() {

    Message msg = ProtonHelper.message();
    msg.setSubject("unauthorized");
    ProtonConnection con = mock(ProtonConnection.class);
    ProtonDelivery delivery = mock(ProtonDelivery.class);
    AuthorizationService authService = mock(AuthorizationService.class);
    when(authService.isAuthorized(any(HonoUser.class), any(ResourceIdentifier.class), anyString())).thenReturn(Future.succeededFuture(Boolean.FALSE));
    Future<Void> processingTracker = Future.future();
    RequestResponseEndpoint<ServiceConfigProperties> endpoint = getEndpoint(true, processingTracker);
    endpoint.setAuthorizationService(authService);

    // WHEN a request for an operation is received that the client is not authorized to invoke
    endpoint.handleMessage(con, receiver, resource, delivery, msg);

    // THEN the the message is rejected
    ArgumentCaptor<DeliveryState> deliveryState = ArgumentCaptor.forClass(DeliveryState.class);
    verify(delivery).disposition(deliveryState.capture(), booleanThat(is(Boolean.TRUE)));
    assertThat(deliveryState.getValue(), instanceOf(Rejected.class));
    verify(receiver, never()).close();
    verify(authService).isAuthorized(Constants.PRINCIPAL_ANONYMOUS, resource, "unauthorized");
    assertFalse(processingTracker.isComplete());
}
 

import io.vertx.proton.ProtonDelivery; //导入依赖的package包/类
/**
 * Verifies that the endpoint processes request messages for operations the client
 * is authorized to invoke.
 */
@Test
public void testHandleMessageProcessesAuthorizedRequests() {

    Message msg = ProtonHelper.message();
    msg.setSubject("get");
    ProtonConnection con = mock(ProtonConnection.class);
    ProtonDelivery delivery = mock(ProtonDelivery.class);
    AuthorizationService authService = mock(AuthorizationService.class);
    when(authService.isAuthorized(any(HonoUser.class), any(ResourceIdentifier.class), anyString())).thenReturn(Future.succeededFuture(Boolean.TRUE));

    Future<Void> processingTracker = Future.future();
    RequestResponseEndpoint<ServiceConfigProperties> endpoint = getEndpoint(true, processingTracker);
    endpoint.setAuthorizationService(authService);

    // WHEN a request for an operation is received that the client is authorized to invoke
    endpoint.handleMessage(con, receiver, resource, delivery, msg);

    // THEN then the message gets processed
    ArgumentCaptor<DeliveryState> deliveryState = ArgumentCaptor.forClass(DeliveryState.class);
    verify(delivery).disposition(deliveryState.capture(), booleanThat(is(Boolean.TRUE)));
    assertThat(deliveryState.getValue(), instanceOf(Accepted.class));
    verify(receiver, never()).close();
    verify(authService).isAuthorized(Constants.PRINCIPAL_ANONYMOUS, resource, "get");
    assertTrue(processingTracker.isComplete());
}
 

import io.vertx.proton.ProtonDelivery; //导入依赖的package包/类
/**
 * Send the AMQP_PUBREL to the related client pubrel address
 *
 * @param amqpPubrelMessage    AMQP_PUBREL message
 */
public void publish(AmqpPubrelMessage amqpPubrelMessage, Handler<AsyncResult<ProtonDelivery>> handler) {

    // send AMQP_PUBREL message

    if (!this.senderPubrel.isOpen()) {

        this.senderPubrel
                .setQoS(ProtonQoS.AT_LEAST_ONCE)
                .open();

        // TODO: think about starting a timer for inactivity on this link for detaching ?
    }

    this.senderPubrel.send(amqpPubrelMessage.toAmqp(), delivery -> {

        if (delivery.getRemoteState() == Accepted.getInstance()) {
            LOG.info("AMQP pubrel delivery {}", delivery.getRemoteState());
            handler.handle(Future.succeededFuture(delivery));
        } else {
            handler.handle(Future.failedFuture(String.format("AMQP pubrel delivery %s", delivery.getRemoteState())));
        }

    });
}