import org.apache.qpid.proton.amqp.messaging.Accepted; //导入依赖的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 org.apache.qpid.proton.amqp.messaging.Accepted; //导入依赖的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 org.apache.qpid.proton.amqp.messaging.Accepted; //导入依赖的package包/类
private void openReceiverWithHandler(ProtonReceiver receiver, RequestHandler requestHandler) {
    receiver.handler((delivery, message) -> {
        vertx.executeBlocking(future -> {
            try {
                Optional<HandlerContext> context = Optional.ofNullable(replyHandlers.get(message.getReplyTo()));
                Message response = requestHandler.handleMessage(message);
                context.ifPresent(ctx -> vertx.runOnContext(v -> ctx.sender.send(response)));
            } catch (Exception e) {
                log.warn("Error handling addressing message", e);
                future.fail(e);
            }
        }, result -> {
            if (result.succeeded()) {
                delivery.disposition(new Accepted(), true);
            } else {
                delivery.disposition(new Rejected(), true);
            }
        });
    });
    receiver.open();
}
 

import org.apache.qpid.proton.amqp.messaging.Accepted; //导入依赖的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())));
        }

    });
}
 

import org.apache.qpid.proton.amqp.messaging.Accepted; //导入依赖的package包/类
@Override
public void onDelivery(Event evt) {
    Delivery dlv = evt.getDelivery();
    Link link = dlv.getLink();
    if (link instanceof Sender) {
        dlv.settle();
    } else {
        Receiver rcv = (Receiver) link;
        if (!dlv.isPartial()) {
            byte[] bytes = new byte[dlv.pending()];
            rcv.recv(bytes, 0, bytes.length);
            String address = router.getAddress(rcv);
            Message message = new Message(bytes);
            messages.put(address, message);
            dlv.disposition(Accepted.getInstance());
            dlv.settle();
            if (!quiet) {
                System.out.println(String.format("Got message(%s): %s", address, message));
            }
            send(address);
        }
    }
}
 

import org.apache.qpid.proton.amqp.messaging.Accepted; //导入依赖的package包/类
@Override
public Accepted readValue() {
    DecoderImpl decoder = getDecoder();
    byte typeCode = decoder.getByteBuffer().get();

    switch (typeCode) {
        case EncodingCodes.LIST0:
            break;
        case EncodingCodes.LIST8:
            decoder.getByteBuffer().get();
            decoder.getByteBuffer().get();
            break;
        case EncodingCodes.LIST32:
            decoder.getByteBuffer().getInt();
            decoder.getByteBuffer().getInt();
            break;
        default:
            throw new DecodeException("Incorrect type found in Accepted type encoding: " + typeCode);
    }

    return Accepted.getInstance();
}
 

import org.apache.qpid.proton.amqp.messaging.Accepted; //导入依赖的package包/类
private void benchmarkDisposition() throws IOException {
    Disposition disposition = new Disposition();
    disposition.setRole(Role.RECEIVER);
    disposition.setSettled(true);
    disposition.setState(Accepted.getInstance());
    disposition.setFirst(UnsignedInteger.valueOf(2));
    disposition.setLast(UnsignedInteger.valueOf(2));

    resultSet.start();
    for (int i = 0; i < ITERATIONS; i++) {
        byteBuf.clear();
        encoder.writeObject(disposition);
    }
    resultSet.encodesComplete();

    resultSet.start();
    for (int i = 0; i < ITERATIONS; i++) {
        byteBuf.flip();
        decoder.readObject();
    }
    resultSet.decodesComplete();

    time("Disposition", resultSet);
}
 

import org.apache.qpid.proton.amqp.messaging.Accepted; //导入依赖的package包/类
@Override
protected Sender createEndpoint(JmsSessionInfo resourceInfo) {
    Coordinator coordinator = new Coordinator();
    coordinator.setCapabilities(TxnCapability.LOCAL_TXN);

    Symbol[] outcomes = new Symbol[]{ Accepted.DESCRIPTOR_SYMBOL, Rejected.DESCRIPTOR_SYMBOL, Released.DESCRIPTOR_SYMBOL, Modified.DESCRIPTOR_SYMBOL };

    Source source = new Source();
    source.setOutcomes(outcomes);

    String coordinatorName = "qpid-jms:coordinator:" + resourceInfo.getId().toString();

    Sender sender = getParent().getSession().getEndpoint().sender(coordinatorName);
    sender.setSource(source);
    sender.setTarget(coordinator);
    sender.setSenderSettleMode(SenderSettleMode.UNSETTLED);
    sender.setReceiverSettleMode(ReceiverSettleMode.FIRST);

    return sender;
}
 

import org.apache.qpid.proton.amqp.messaging.Accepted; //导入依赖的package包/类
/**
 * Verifies that telemetry data is discarded if no downstream credit is available.
 * 
 * @param ctx The test context.
 */
@SuppressWarnings("unchecked")
@Test
public void testProcessMessageDiscardsMessageIfNoCreditIsAvailable(final TestContext ctx) {

    final UpstreamReceiver client = newClient();
    final ProtonDelivery delivery = mock(ProtonDelivery.class);
    when(delivery.remotelySettled()).thenReturn(Boolean.FALSE);

    // GIVEN an adapter with a connection to a downstream container
    ProtonSender sender = newMockSender(false);
    when(sender.sendQueueFull()).thenReturn(true);
    ForwardingEventDownstreamAdapter adapter = new ForwardingEventDownstreamAdapter(vertx, newMockSenderFactory(sender));
    adapter.setMetrics(mock(MessagingMetrics.class));
    adapter.setDownstreamConnectionFactory(newMockConnectionFactory(false));
    adapter.start(Future.future());
    adapter.addSender(client, sender);

    // WHEN processing an event
    Message msg = ProtonHelper.message(TELEMETRY_MSG_CONTENT);
    MessageHelper.addDeviceId(msg, DEVICE_ID);
    adapter.processMessage(client, delivery, msg);

    // THEN the the message is accepted
    verify(delivery).disposition(any(Accepted.class), eq(Boolean.TRUE));
    // but is not delivered to the downstream container
    verify(sender, never()).send(any(Message.class), any(Handler.class));
}
 

import org.apache.qpid.proton.amqp.messaging.Accepted; //导入依赖的package包/类
/**
 * Verifies that an event uploaded by an upstream client is forwarded to the
 * downstream container.
 * 
 * @param ctx The test context.
 */
@SuppressWarnings("unchecked")
@Test
public void testProcessMessageForwardsMessageToDownstreamSender(final TestContext ctx) {

    final UpstreamReceiver client = newClient();
    final ProtonDelivery delivery = mock(ProtonDelivery.class);
    final ProtonDelivery downstreamDelivery = mock(ProtonDelivery.class);
    when(downstreamDelivery.getRemoteState()).thenReturn(ACCEPTED);
    when(downstreamDelivery.remotelySettled()).thenReturn(true);

    // GIVEN an adapter with a connection to a downstream container
    final Async msgSent = ctx.async();
    ProtonSender sender = newMockSender(false);
    when(sender.send(any(Message.class), any(Handler.class))).then(invocation -> {
        msgSent.complete();
        invocation.getArgumentAt(1, Handler.class).handle(downstreamDelivery);
        return null;
    });
    ForwardingEventDownstreamAdapter adapter = new ForwardingEventDownstreamAdapter(vertx, newMockSenderFactory(sender));
    adapter.setMetrics(mock(MessagingMetrics.class));
    adapter.setDownstreamConnectionFactory(newMockConnectionFactory(false));
    adapter.start(Future.future());
    adapter.addSender(client, sender);

    // WHEN processing an event
    Message msg = ProtonHelper.message(EVENT_MSG_CONTENT);
    MessageHelper.addDeviceId(msg, DEVICE_ID);
    adapter.processMessage(client, delivery, msg);

    // THEN the message has been delivered to the downstream container
    msgSent.await(1000);
    // and disposition was returned
    verify(delivery).disposition(any(Accepted.class), eq(Boolean.TRUE));
}
 

import org.apache.qpid.proton.amqp.messaging.Accepted; //导入依赖的package包/类
@Override
protected Future<ProtonDelivery> sendMessage(final Message message) {

    Objects.requireNonNull(message);

    final String messageId = String.format("%s-%d", getClass().getSimpleName(), MESSAGE_COUNTER.getAndIncrement());
    message.setMessageId(messageId);
    final ProtonDelivery result = sender.send(message, deliveryUpdated -> {

        if (deliveryUpdated.remotelySettled()) {
            if (Accepted.class.isInstance(deliveryUpdated.getRemoteState())) {
                LOG.trace("message [message ID: {}] accepted by peer", messageId);
            } else if (Rejected.class.isInstance(deliveryUpdated.getRemoteState())) {
                Rejected remoteState = (Rejected) deliveryUpdated.getRemoteState();
                if (remoteState.getError() == null) {
                    LOG.debug("message [message ID: {}] rejected by peer", messageId);
                } else {
                    LOG.debug("message [message ID: {}] rejected by peer: {}, {}", messageId,
                            remoteState.getError().getCondition(), remoteState.getError().getDescription());
                }
            } else {
                LOG.debug("message [message ID: {}] not accepted by peer: {}", messageId, deliveryUpdated.getRemoteState());
            }
        } else {
            LOG.warn("peer did not settle telemetry message [message ID: {}, remote state: {}]", messageId, deliveryUpdated.getRemoteState());
        }
    });
    LOG.trace("sent telemetry message [ID: {}], remaining credit: {}, queued messages: {}", messageId, sender.getCredit(), sender.getQueued());

    return Future.succeededFuture(result);
}
 

import org.apache.qpid.proton.amqp.messaging.Accepted; //导入依赖的package包/类
@Override
protected Future<ProtonDelivery> sendMessage(final Message message) {

    Objects.requireNonNull(message);

    final Future<ProtonDelivery> result = Future.future();
    final String messageId = String.format("%s-%d", getClass().getSimpleName(), MESSAGE_COUNTER.getAndIncrement());
    message.setMessageId(messageId);
    sender.send(message, deliveryUpdated -> {

        if (deliveryUpdated.remotelySettled()) {
            if (Accepted.class.isInstance(deliveryUpdated.getRemoteState())) {
                LOG.trace("event [message ID: {}] accepted by peer", messageId);
                result.complete(deliveryUpdated);
            } else if (Rejected.class.isInstance(deliveryUpdated.getRemoteState())) {
                Rejected rejected = (Rejected) deliveryUpdated.getRemoteState();
                if (rejected.getError() == null) {
                    LOG.debug("event [message ID: {}] rejected by peer", messageId);
                    result.fail(new ClientErrorException(HttpURLConnection.HTTP_BAD_REQUEST));
                } else {
                    LOG.debug("event [message ID: {}] rejected by peer: {}, {}", messageId,
                            rejected.getError().getCondition(), rejected.getError().getDescription());
                    result.fail(new ClientErrorException(HttpURLConnection.HTTP_BAD_REQUEST, rejected.getError().getDescription()));
                }
            } else {
                LOG.debug("event [message ID: {}] not accepted by peer: {}", messageId, deliveryUpdated.getRemoteState());
                result.fail(new ClientErrorException(HttpURLConnection.HTTP_BAD_REQUEST));
            }
        } else {
            LOG.warn("peer did not settle event, failing delivery [new remote state: {}]", deliveryUpdated.getRemoteState());
            result.fail(new ServerErrorException(HttpURLConnection.HTTP_INTERNAL_ERROR));
        }
    });
    LOG.trace("sent event [ID: {}], remaining credit: {}, queued messages: {}", messageId, sender.getCredit(), sender.getQueued());

    return result;
}
 

import org.apache.qpid.proton.amqp.messaging.Accepted; //导入依赖的package包/类
/**
 * Verifies that the sender waits for the peer to settle and
 * accept a message before succeeding the returned future.
 * 
 * @param ctx The vert.x test context.
 */
@SuppressWarnings({ "unchecked" })
@Test
public void testSendMessageWaitsForAcceptedOutcome(final TestContext ctx) {

    // GIVEN a sender that has credit
    when(sender.sendQueueFull()).thenReturn(Boolean.FALSE);
    MessageSender messageSender = new EventSenderImpl(config, sender, "tenant", "telemetry/tenant", context);
    final AtomicReference<Handler<ProtonDelivery>> handlerRef = new AtomicReference<>();
    doAnswer(invocation -> {
        handlerRef.set(invocation.getArgumentAt(1, Handler.class));
        return mock(ProtonDelivery.class);
    }).when(sender).send(any(Message.class), any(Handler.class));

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

    // THEN the message has been sent
    // and the result is not completed yet
    verify(sender).send(any(Message.class), eq(handlerRef.get()));
    assertFalse(result.isComplete());

    // until it gets accepted by the peer
    ProtonDelivery accepted = mock(ProtonDelivery.class);
    when(accepted.remotelySettled()).thenReturn(Boolean.TRUE);
    when(accepted.getRemoteState()).thenReturn(new Accepted());
    handlerRef.get().handle(accepted);

    assertTrue(result.succeeded());
}
 

import org.apache.qpid.proton.amqp.messaging.Accepted; //导入依赖的package包/类
private void sendNext(ProtonConnection connection, ProtonSender sender) {

        Message message;
        if (messageQueue.hasNext() && (message = messageQueue.next()) != null) {
            if (sender.getQoS().equals(ProtonQoS.AT_MOST_ONCE)) {
                sender.send(message);
                numSent.incrementAndGet();
                if(predicate.test(message)) {
                    promise.complete(numSent.get());
                } else {
                    vertx.runOnContext(id -> sendNext(connection, sender));
                }
            } else {
                sender.send(message, protonDelivery -> {
                    if (protonDelivery.getRemoteState().equals(Accepted.getInstance())) {
                        numSent.incrementAndGet();
                        if(predicate.test(message)) {
                            promise.complete(numSent.get());
                            connection.close();
                        } else {
                            sendNext(connection, sender);
                        }
                    } else {
                        promise.completeExceptionally(new IllegalStateException("Message not accepted (remote state: "+protonDelivery.getRemoteState()+") after " + numSent.get() + " messages sent"));
                        connection.close();
                    }
                });
            }
        } else {
            if(predicate.test(null)) {
                promise.complete(numSent.get());
            } else {
                promise.completeExceptionally(new RuntimeException("No more messages to send after + " + numSent.get() + " messages sent"));
            }
            connection.close();
        }
    }
 

import org.apache.qpid.proton.amqp.messaging.Accepted; //导入依赖的package包/类
private void connectionOpened(ProtonConnection conn, String linkName, Source source) {
    ProtonReceiver receiver = conn.createReceiver(source.getAddress(), new ProtonLinkOptions().setLinkName(linkName));
    receiver.setSource(source);
    receiver.setPrefetch(0);
    receiver.handler((protonDelivery, message) -> {
        messages.add(message);
        protonDelivery.disposition(Accepted.getInstance(), true);
        if (done.test(message)) {
            promise.complete(messages);
            conn.close();
        } else {
            receiver.flow(1);
        }
    });
    receiver.openHandler(result -> {
        Logging.log.info("Receiver link '" + source.getAddress() + "' opened, granting credits");
        receiver.flow(1);
        connectLatch.countDown();
    });

    receiver.closeHandler(closed -> {
        if (receiver.getRemoteCondition() != null && LinkError.REDIRECT.equals(receiver.getRemoteCondition().getCondition())) {
            String relocated = (String) receiver.getRemoteCondition().getInfo().get("address");
            Logging.log.info("Receiver link redirected to '" + relocated + "'");
            Source newSource = linkOptions.getSource();
            newSource.setAddress(relocated);
            String newLinkName = linkOptions.getLinkName().orElse(newSource.getAddress());

            vertx.runOnContext(id -> connectionOpened(conn, newLinkName, newSource));
        } else {
            handleError(conn, receiver.getRemoteCondition());
        }
        receiver.close();
    });
    receiver.open();
}
 

import org.apache.qpid.proton.amqp.messaging.Accepted; //导入依赖的package包/类
/**
 * Send the AMQP_WILL message to the Last Will and Testament Service
 *
 * @param amqpWillMessage   AMQP_WILL message
 * @param handler   callback called on message delivered
 */
public void sendWill(AmqpWillMessage amqpWillMessage, Handler<AsyncResult<ProtonDelivery>> handler) {

    // send AMQP_WILL message with will information
    this.sender.send(amqpWillMessage.toAmqp(), delivery -> {

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

import org.apache.qpid.proton.amqp.messaging.Accepted; //导入依赖的package包/类
/**
 * Settle the delivery for a received message
 *
 * @param messageId message identifier to settle
 */
public void settle(Object messageId) {

    if (this.deliveries.containsKey(messageId)) {
        ProtonDelivery delivery = this.deliveries.remove(messageId);
        delivery.disposition(Accepted.getInstance(), true);
    }
}
 

import org.apache.qpid.proton.amqp.messaging.Accepted; //导入依赖的package包/类
/**
 * Send the AMQP_LIST message to the Subscription Service
 *
 * @param amqpSessionMessage    AMQP_LIST message
 * @param handler   callback called on message delivered
 */
public void sendList(AmqpListMessage amqpSessionMessage, Handler<AsyncResult<ProtonDelivery>> handler) {

    // send AMQP_LIST message
    this.sender.send(amqpSessionMessage.toAmqp(), delivery -> {

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

import org.apache.qpid.proton.amqp.messaging.Accepted; //导入依赖的package包/类
/**
 * Send the AMQP_CLOSE message to the Subscription Service
 *
 * @param amqpCloseMessage  AMQP_CLOSE message
 * @param handler   callback called on message delivered
 */
public void sendClose(AmqpCloseMessage amqpCloseMessage, Handler<AsyncResult<ProtonDelivery>> handler) {

    // send AMQP_CLOSE message
    this.sender.send(amqpCloseMessage.toAmqp(), delivery -> {

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

import org.apache.qpid.proton.amqp.messaging.Accepted; //导入依赖的package包/类
/**
 * Send the AMQP_SUBSCRIBE message to the Subscription Service
 *
 * @param amqpSubscribeMessage  AMQP_SUBSCRIBE message
 * @param handler   callback called on message delivered
 */
public void sendSubscribe(AmqpSubscribeMessage amqpSubscribeMessage, Handler<AsyncResult<ProtonDelivery>> handler) {

    // send AMQP_SUBSCRIBE message
    this.sender.send(amqpSubscribeMessage.toAmqp(), delivery -> {

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