import org.apache.qpid.proton.engine.Delivery; //导入依赖的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.engine.Delivery; //导入依赖的package包/类
@Override
public void onDelivery(Event evt) {
    Delivery dlv = evt.getDelivery();
    if (dlv.getLink() instanceof Receiver) {
        Receiver receiver = (Receiver) dlv.getLink();

        if (!dlv.isPartial()) {
            byte[] bytes = new byte[dlv.pending()];
            receiver.recv(bytes, 0, bytes.length);
            Message msg = new Message(bytes);

            if (!quiet) {
                System.out.println(String.format("Got message: %s", msg));
            }
            received++;
            dlv.settle();
        }

        if ((received >= count) || (!block && !receiver.draining())) {
            receiver.getSession().getConnection().close();
        }
    }
}
 

import org.apache.qpid.proton.engine.Delivery; //导入依赖的package包/类
@Override
public void onLinkFlow(Event evt) {
    Link link = evt.getLink();
    if (link instanceof Sender) {
        Sender sender = (Sender) link;
        while ((sent < count) && sender.getCredit() > 0) {
            Delivery dlv = sender.delivery(String.format("spout-%s", sent).getBytes());

            Message msg = new Message(String.format("Hello World! [%s]", sent));
            byte[] bytes = msg.getBytes();
            sender.send(bytes, 0, bytes.length);
            sender.advance();

            if (!quiet) {
                System.out.println(String.format("Sent %s to %s: %s", new String(dlv.getTag()),
                                                 sender.getTarget().getAddress(), msg));
            }
            sent++;
        }
    }
}
 

import org.apache.qpid.proton.engine.Delivery; //导入依赖的package包/类
@Override
public void onLinkFlow(Event event) {
    Sender snd = (Sender)event.getLink();
    if (snd.getCredit() > 0) {
        byte[] msgData = new byte[1024];
        int length;
        while(true) {
            try {
                length = message.encode(msgData, 0, msgData.length);
                break;
            } catch(BufferOverflowException e) {
                msgData = new byte[msgData.length * 2];
            }
        }
        byte[] tag = String.valueOf(nextTag++).getBytes();
        Delivery dlv = snd.delivery(tag);
        snd.send(msgData, 0, length);
        dlv.settle();
        snd.advance();
        snd.close();
        snd.getSession().close();
        snd.getSession().getConnection().close();
    }
}
 

import org.apache.qpid.proton.engine.Delivery; //导入依赖的package包/类
@Override
public Reactor getReactor() {
    if (context instanceof Reactor) {
        return (Reactor) context;
    } else if (context instanceof Task) {
        return ((Task)context).getReactor();
    } else if (context instanceof Transport) {
        return ((TransportImpl)context).getReactor();
    } else if (context instanceof Delivery) {
        return ((Delivery)context).getLink().getSession().getConnection().getReactor();
    } else if (context instanceof Link) {
        return ((Link)context).getSession().getConnection().getReactor();
    } else if (context instanceof Session) {
        return ((Session)context).getConnection().getReactor();
    } else if (context instanceof Connection) {
        return ((Connection)context).getReactor();
    } else if (context instanceof Selectable) {
        return ((Selectable)context).getReactor();
    }
    return null;
}
 

import org.apache.qpid.proton.engine.Delivery; //导入依赖的package包/类
private Delivery sendMessageToClient(String deliveryTag, int messageBody)
{
    byte[] tag = deliveryTag.getBytes(StandardCharsets.UTF_8);

    Message m = Proton.message();
    m.setBody(new AmqpValue(messageBody));

    byte[] encoded = new byte[BUFFER_SIZE];
    int len = m.encode(encoded, 0, BUFFER_SIZE);

    assertTrue("given array was too small", len < BUFFER_SIZE);

    Sender serverSender = getServer().getSender();
    Delivery serverDelivery = serverSender.delivery(tag);
    int sent = serverSender.send(encoded, 0, len);

    assertEquals("sender unable to send all data at once as assumed for simplicity", len, sent);

    boolean senderAdvanced = serverSender.advance();
    assertTrue("sender has not advanced", senderAdvanced);

    return serverDelivery;
}
 

import org.apache.qpid.proton.engine.Delivery; //导入依赖的package包/类
@Test
public void testGetTransportWithDeliveryContext()
{
    Transport transport = Transport.Factory.create();
    Connection connection = Connection.Factory.create();
    transport.bind(connection);

    Session session = connection.session();
    Sender sender = session.sender("mySender");

    Delivery delivery = sender.delivery("tag".getBytes());

    EventImpl event = createEvent(delivery, Event.Type.DELIVERY);

    assertNotNull("No transport returned", event.getTransport());
    assertSame("Incorrect transport returned", transport, event.getTransport());
}
 

import org.apache.qpid.proton.engine.Delivery; //导入依赖的package包/类
/**
 * Creates a new AmqpMessage that wraps the information necessary to handle
 * an incoming delivery.
 *
 * @param receiver the AmqpReceiver that received this message.
 * @param message  the Proton message that was received.
 * @param delivery the Delivery instance that produced this message.
 */
@SuppressWarnings("unchecked")
public AmqpMessage(AmqpReceiver receiver, Message message, Delivery delivery) {
   this.receiver = receiver;
   this.message = message;
   this.delivery = delivery;

   if (message.getMessageAnnotations() != null) {
      messageAnnotationsMap = message.getMessageAnnotations().getValue();
   }

   if (message.getApplicationProperties() != null) {
      applicationPropertiesMap = message.getApplicationProperties().getValue();
   }

   if (message.getDeliveryAnnotations() != null) {
      deliveryAnnotationsMap = message.getDeliveryAnnotations().getValue();
   }
}
 

import org.apache.qpid.proton.engine.Delivery; //导入依赖的package包/类
private void processDelivery(Delivery incoming) throws Exception {
   doDeliveryInspection(incoming);

   Message message = null;
   try {
      message = decodeIncomingMessage(incoming);
   } catch (Exception e) {
      LOG.warn("Error on transform: {}", e.getMessage());
      deliveryFailed(incoming, true);
      return;
   }

   AmqpMessage amqpMessage = new AmqpMessage(this, message, incoming);
   // Store reference to envelope in delivery context for recovery
   incoming.setContext(amqpMessage);
   prefetch.add(amqpMessage);

   // We processed a message, signal completion
   // of a message pull request if there is one.
   if (pullRequest != null) {
      pullRequest.onSuccess();
      pullRequest = null;
   }
}
 

import org.apache.qpid.proton.engine.Delivery; //导入依赖的package包/类
protected Message decodeIncomingMessage(Delivery incoming) {
   int count;

   byte[] chunk = new byte[2048];
   ByteArrayOutputStream stream = new ByteArrayOutputStream();

   while ((count = getEndpoint().recv(chunk, 0, chunk.length)) > 0) {
      stream.write(chunk, 0, count);
   }

   byte[] messageBytes = stream.toByteArray();

   try {
      Message protonMessage = Message.Factory.create();
      protonMessage.decode(messageBytes, 0, messageBytes.length);
      return protonMessage;
   } finally {
      try {
         stream.close();
      } catch (IOException e) {
      }
   }
}
 

import org.apache.qpid.proton.engine.Delivery; //导入依赖的package包/类
private void encodeAndSend(Message message, Delivery delivery) throws IOException {

      int encodedSize;
      while (true) {
         try {
            encodedSize = message.encode(encodeBuffer, 0, encodeBuffer.length);
            break;
         } catch (java.nio.BufferOverflowException e) {
            encodeBuffer = new byte[encodeBuffer.length * 2];
         }
      }

      int sentSoFar = 0;

      while (true) {
         int sent = getEndpoint().send(encodeBuffer, sentSoFar, encodedSize - sentSoFar);
         if (sent > 0) {
            sentSoFar += sent;
            if ((encodedSize - sentSoFar) == 0) {
               break;
            }
         } else {
            LOG.warn("{} failed to send any data from current Message.", this);
         }
      }
   }
 

import org.apache.qpid.proton.engine.Delivery; //导入依赖的package包/类
@Override
public void rollbackRedelivery(Transaction txn, MessageReference ref, long timeBase, Map<QueueImpl, LinkedList<MessageReference>> queueMap) throws Exception {
   ProtonTransactionImpl tx = (ProtonTransactionImpl) txn;

   if (tx.getDeliveries().containsKey(ref)) {
      Delivery del = tx.getDeliveries().get(ref).getA();
      ServerConsumer consumer = (ServerConsumer) tx.getDeliveries().get(ref).getB().getBrokerConsumer();
      // Rollback normally if the delivery is not settled or a forced TX rollback is done (e.g. connection drop).
      if (del.remotelySettled() || !tx.isDischarged()) {
         super.rollbackRedelivery(tx, ref, timeBase, queueMap);
      } else {
         ref.incrementDeliveryCount();
         consumer.backToDelivering(ref);
         del.disposition(del.getLocalState() == null ? del.getDefaultDeliveryState() : del.getLocalState());
      }
   } else {
      super.rollbackRedelivery(tx, ref, timeBase, queueMap);
   }
}
 

import org.apache.qpid.proton.engine.Delivery; //导入依赖的package包/类
void disposition(long sequence, DeliveryState state, int... flags) {
  int flag = flags.length == 1 ? flags[0] : 0;
  boolean cumilative = (flag & CUMULATIVE) != 0;
  boolean settle = (flag & SETTLE) != 0;

  long count = cumilative ? _lastDispositionMark.get() : sequence;
  long end = sequence;

  while (count <= end) {
    Delivery d = _unsettled.get(count);
    if (d != null) {
      d.disposition(state);
    }
    count++;
  }
  _lastDispositionMark.set(end);
  if (settle) {
    settle(sequence, cumilative, false);
  }
  _conn.write();
}
 

import org.apache.qpid.proton.engine.Delivery; //导入依赖的package包/类
void settle(long sequence, boolean cumilative, boolean write) {
  long count = cumilative ? _lastSettled.get() : sequence;
  long end = sequence;

  while (count <= end) {
    Delivery d = _unsettled.get(count);
    if (d != null) {
      if (!d.isSettled() && d.getLink().getReceiverSettleMode() == ReceiverSettleMode.FIRST) {
        d.settle();
        ((IncomingLinkImpl) d.getLink().getContext()).decrementUnsettledCount();
        _unsettled.remove(count);
      }
    }
    count++;
  }
  _lastSettled.set(end);
  _conn.write();
}
 

import org.apache.qpid.proton.engine.Delivery; //导入依赖的package包/类
TrackerImpl send(org.apache.qpid.proton.message.Message m) throws MessageFormatException, MessagingException {
  checkClosed();
  Sender sender = (Sender) _link;
  byte[] tag = longToBytes(_ssn.getNextDeliveryTag());
  Delivery delivery = sender.delivery(tag);
  TrackerImpl tracker = new TrackerImpl(_ssn);
  delivery.setContext(tracker);

  if (m.getAddress() == null) {
    m.setAddress(_address);
  }
  byte[] buffer = new byte[1024];
  int encoded = m.encode(buffer, 0, buffer.length);
  sender.send(buffer, 0, encoded);
  if (sender.getSenderSettleMode() == SenderSettleMode.SETTLED) {
    delivery.settle();
    tracker.markSettled();
  }
  sender.advance();
  _ssn.getConnection().write();
  return tracker;
}
 

import org.apache.qpid.proton.engine.Delivery; //导入依赖的package包/类
void disposition(InboundMessage msg, DeliveryState state, int... flags) throws NetworkException
{
    int flag = flags.length == 1 ? flags[0] : 0;
    boolean cumilative = (flag & CUMULATIVE) != 0;
    boolean settle = (flag & SETTLE) != 0;

    long count = cumilative ? _lastDispositionMark.get() : msg.getSequence();
    long end = msg.getSequence();

    while (count <= end)
    {
        Delivery d = _unsettled.get(count);
        if (d != null)
        {
            d.disposition(state);
        }
        count++;
    }
    _lastDispositionMark.set(end);
    if (settle)
    {
        settle(msg, cumilative, false);
    }
    _conn.write();
}
 

import org.apache.qpid.proton.engine.Delivery; //导入依赖的package包/类
void settle(InboundMessage msg, boolean cumilative, boolean write) throws NetworkException
{
    long count = cumilative ? _lastSettled.get() : msg.getSequence();
    long end = msg.getSequence();

    while (count <= end)
    {
        Delivery d = _unsettled.get(count);
        if (d != null)
        {
            if (!d.isSettled() && d.getLink().getReceiverSettleMode() == ReceiverSettleMode.FIRST)
            {
                d.settle();
                ((InboundLinkImpl) d.getLink().getContext()).decrementUnsettledCount();
                _unsettled.remove(count);
            }
        }
        count++;
    }
    _lastSettled.set(end);
    _conn.write();
}
 

import org.apache.qpid.proton.engine.Delivery; //导入依赖的package包/类
protected ByteBuf unwrapIncomingMessage(Delivery incoming) {
    int count;

    // Attempt to preemptively size the buffer for the incoming delivery.
    if (incomingBuffer.capacity() < incoming.available()) {
        incomingBuffer.capacity(incoming.available());
    }

    while ((count = getEndpoint().recv(incomingBuffer.array(), incomingBuffer.writerIndex(), incomingBuffer.writableBytes())) > 0) {
        incomingBuffer.writerIndex(incomingBuffer.writerIndex() + count);
        if (!incomingBuffer.isWritable()) {
            incomingBuffer.capacity((int) (incomingBuffer.capacity() * 1.5));
        }
    }

    return incomingBuffer;
}
 

import org.apache.qpid.proton.engine.Delivery; //导入依赖的package包/类
private void releasePrefetch() {
    Delivery delivery = getEndpoint().head();

    while (delivery != null) {
        Delivery current = delivery;
        delivery = delivery.next();

        if (!(current.getContext() instanceof JmsInboundMessageDispatch)) {
            LOG.debug("{} Found incomplete delivery with no context during release processing", AmqpConsumer.this);
            continue;
        }

        JmsInboundMessageDispatch envelope = (JmsInboundMessageDispatch) current.getContext();
        if (!envelope.isDelivered()) {
            current.disposition(Released.getInstance());
            current.settle();
        }
    }
}
 

import org.apache.qpid.proton.engine.Delivery; //导入依赖的package包/类
protected EncodedMessage readIncomingMessage(Delivery incoming) {
    Buffer buffer;
    int count;

    while ((count = endpoint.recv(incomingBuffer, 0, incomingBuffer.length)) > 0) {
        streamBuffer.write(incomingBuffer, 0, count);
    }

    buffer = streamBuffer.toBuffer();

    try {
        return new EncodedMessage(incoming.getMessageFormat(), buffer.data, buffer.offset, buffer.length);
    } finally {
        streamBuffer.reset();
    }
}