import org.apache.qpid.proton.amqp.transport.ReceiverSettleMode; //导入依赖的package包/类
@Test
public void testEncodeDecodeTransfers() {
    Transfer transfer = new Transfer();
    transfer.setHandle(UnsignedInteger.ONE);
    transfer.setDeliveryTag(new Binary(new byte[] {0, 1}));
    transfer.setMessageFormat(UnsignedInteger.ZERO);
    transfer.setDeliveryId(UnsignedInteger.valueOf(127));
    transfer.setAborted(false);
    transfer.setBatchable(true);
    transfer.setRcvSettleMode(ReceiverSettleMode.SECOND);

    encoder.writeObject(transfer);
    buffer.clear();
    final Transfer outputValue = (Transfer) decoder.readObject();

    assertEquals(transfer.getHandle(), outputValue.getHandle());
    assertEquals(transfer.getMessageFormat(), outputValue.getMessageFormat());
    assertEquals(transfer.getDeliveryTag(), outputValue.getDeliveryTag());
    assertEquals(transfer.getDeliveryId(), outputValue.getDeliveryId());
    assertEquals(transfer.getAborted(), outputValue.getAborted());
    assertEquals(transfer.getBatchable(), outputValue.getBatchable());
    assertEquals(transfer.getRcvSettleMode(), outputValue.getRcvSettleMode());
}
 

import org.apache.qpid.proton.amqp.transport.ReceiverSettleMode; //导入依赖的package包/类
@Test
public void testSkipValue() {
    Transfer transfer = new Transfer();
    transfer.setHandle(UnsignedInteger.ONE);
    transfer.setDeliveryTag(new Binary(new byte[] {0, 1}));
    transfer.setMessageFormat(UnsignedInteger.ZERO);
    transfer.setDeliveryId(UnsignedInteger.valueOf(127));
    transfer.setAborted(false);
    transfer.setBatchable(true);
    transfer.setRcvSettleMode(ReceiverSettleMode.SECOND);

    encoder.writeObject(transfer);

    transfer.setHandle(UnsignedInteger.valueOf(2));

    encoder.writeObject(transfer);

    buffer.clear();

    TypeConstructor<?> type = decoder.readConstructor();
    assertEquals(Transfer.class, type.getTypeClass());
    type.skipValue();

    Transfer result = (Transfer) decoder.readObject();
    assertEquals(UnsignedInteger.valueOf(2), result.getHandle());
}
 

import org.apache.qpid.proton.amqp.transport.ReceiverSettleMode; //导入依赖的package包/类
public void doTestSenderSettlementModeIsHonored(SenderSettleMode settleMode) throws Exception {
   AmqpClient client = createAmqpClient();
   AmqpConnection connection = addConnection(client.connect());
   AmqpSession session = connection.createSession();

   AmqpReceiver receiver = session.createReceiver("queue://" + getTestName(), settleMode, ReceiverSettleMode.FIRST);

   Queue queueView = getProxyToQueue(getQueueName());
   assertNotNull(queueView);
   assertEquals(0, queueView.getMessageCount());
   assertEquals(1, server.getTotalConsumerCount());

   assertEquals(settleMode, receiver.getEndpoint().getRemoteSenderSettleMode());

   receiver.close();

   connection.close();
}
 

import org.apache.qpid.proton.amqp.transport.ReceiverSettleMode; //导入依赖的package包/类
private void doTestReceiverSettlementModeForcedToFirst(ReceiverSettleMode modeToUse) throws Exception {
   AmqpClient client = createAmqpClient();
   AmqpConnection connection = addConnection(client.connect());
   AmqpSession session = connection.createSession();

   AmqpReceiver receiver = session.createReceiver("queue://" + getTestName(), SenderSettleMode.MIXED, modeToUse);

   Queue queueView = getProxyToQueue(getQueueName());
   assertNotNull(queueView);
   assertEquals(0, queueView.getMessageCount());
   assertEquals(1, server.getTotalConsumerCount());

   assertEquals(ReceiverSettleMode.FIRST, receiver.getEndpoint().getRemoteReceiverSettleMode());

   receiver.close();
   connection.close();
}
 

import org.apache.qpid.proton.amqp.transport.ReceiverSettleMode; //导入依赖的package包/类
private void doTestReceiverSettlementModeForcedToFirst(ReceiverSettleMode modeToUse) throws Exception {
   AmqpClient client = createAmqpClient();
   AmqpConnection connection = addConnection(client.connect());
   AmqpSession session = connection.createSession();

   AmqpSender sender = session.createSender("queue://" + getTestName(), SenderSettleMode.UNSETTLED, modeToUse);

   Queue queueView = getProxyToQueue(getQueueName());
   assertNotNull(queueView);
   assertEquals(0, queueView.getMessageCount());

   assertEquals(ReceiverSettleMode.FIRST, sender.getEndpoint().getRemoteReceiverSettleMode());

   sender.close();

   connection.close();
}
 

import org.apache.qpid.proton.amqp.transport.ReceiverSettleMode; //导入依赖的package包/类
@Override
protected void doOpen() {
   Coordinator coordinator = new Coordinator();
   coordinator.setCapabilities(TxnCapability.LOCAL_TXN);
   Source source = new Source();

   String coordinatorName = "qpid-jms:coordinator:" + session.getConnection().getConnectionId();

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

   setEndpoint(sender);

   super.doOpen();
}
 

import org.apache.qpid.proton.amqp.transport.ReceiverSettleMode; //导入依赖的package包/类
/**
 * Create a sender instance using the given address
 *
 * @param address
 *        the address to which the sender will produce its messages.
 * @param senderSettlementMode
 *        controls the settlement mode used by the created Sender
 * @param receiverSettlementMode
 *        controls the desired settlement mode used by the remote Receiver
 *
 * @return a newly created sender that is ready for use.
 *
 * @throws Exception if an error occurs while creating the sender.
 */
public AmqpSender createSender(final String address, final SenderSettleMode senderMode, ReceiverSettleMode receiverMode) throws Exception {
   checkClosed();

   final AmqpSender sender = new AmqpSender(AmqpSession.this, address, getNextSenderId(), senderMode, receiverMode);
   final ClientFuture request = new ClientFuture();

   connection.getScheduler().execute(new Runnable() {

      @Override
      public void run() {
         checkClosed();
         sender.setStateInspector(getStateInspector());
         sender.open(request);
         pumpToProtonTransport(request);
      }
   });

   request.sync();

   return sender;
}
 

import org.apache.qpid.proton.amqp.transport.ReceiverSettleMode; //导入依赖的package包/类
/**
 * Create a receiver instance using the given address
 *
 * @param address
 *        the address to which the receiver will subscribe for its messages.
 * @param senderSettlementMode
 *        controls the desired settlement mode used by the remote Sender
 * @param receiverSettlementMode
 *        controls the settlement mode used by the created Receiver
 *
 * @return a newly created receiver that is ready for use.
 *
 * @throws Exception if an error occurs while creating the receiver.
 */
public AmqpReceiver createReceiver(String address, SenderSettleMode senderMode, ReceiverSettleMode receiverMode) throws Exception {
   checkClosed();

   final ClientFuture request = new ClientFuture();
   final AmqpReceiver receiver = new AmqpReceiver(AmqpSession.this, address, getNextReceiverId(), senderMode, receiverMode);

   connection.getScheduler().execute(new Runnable() {

      @Override
      public void run() {
         checkClosed();
         receiver.setStateInspector(getStateInspector());
         receiver.open(request);
         pumpToProtonTransport(request);
      }
   });

   request.sync();

   return receiver;
}
 

import org.apache.qpid.proton.amqp.transport.ReceiverSettleMode; //导入依赖的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.amqp.transport.ReceiverSettleMode; //导入依赖的package包/类
@Override
public T setQoS(ProtonQoS qos) {
  if (qos == null) {
    throw new IllegalArgumentException("Value must be specified");
  }
  switch (qos) {
  case AT_MOST_ONCE:
    link.setSenderSettleMode(SenderSettleMode.SETTLED);
    link.setReceiverSettleMode(ReceiverSettleMode.FIRST);
    break;
  case AT_LEAST_ONCE:
    link.setSenderSettleMode(SenderSettleMode.UNSETTLED);
    link.setReceiverSettleMode(ReceiverSettleMode.FIRST);
    break;
  }
  return self();
}
 

import org.apache.qpid.proton.amqp.transport.ReceiverSettleMode; //导入依赖的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.amqp.transport.ReceiverSettleMode; //导入依赖的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.transport.ReceiverSettleMode; //导入依赖的package包/类
@Override
protected void doOpen() {

    String sourceAddress = info.getName();
    if (info.isQueue()) {
        sourceAddress = connection.getTempQueuePrefix() + sourceAddress;
    } else {
        sourceAddress = connection.getTempQueuePrefix() + sourceAddress;
    }
    Source source = new Source();
    source.setAddress(sourceAddress);
    Target target = new Target();
    target.setDynamic(true);

    String senderName = sourceAddress;
    endpoint = session.getProtonSession().sender(senderName);
    endpoint.setSource(source);
    endpoint.setTarget(target);
    endpoint.setSenderSettleMode(SenderSettleMode.UNSETTLED);
    endpoint.setReceiverSettleMode(ReceiverSettleMode.FIRST);

    this.connection.addTemporaryDestination(this);
}
 

import org.apache.qpid.proton.amqp.transport.ReceiverSettleMode; //导入依赖的package包/类
@Override
protected void doOpen() {
    JmsDestination destination = info.getDestination();

    String destnationName = session.getQualifiedName(destination);
    String sourceAddress = getProducerId().toString();
    Source source = new Source();
    source.setAddress(sourceAddress);
    Target target = new Target();
    target.setAddress(destnationName);

    String senderName = sourceAddress + ":" + destnationName != null ? destnationName : "Anonymous";
    endpoint = session.getProtonSession().sender(senderName);
    endpoint.setSource(source);
    endpoint.setTarget(target);
    if (presettle) {
        endpoint.setSenderSettleMode(SenderSettleMode.SETTLED);
    } else {
        endpoint.setSenderSettleMode(SenderSettleMode.UNSETTLED);
    }
    endpoint.setReceiverSettleMode(ReceiverSettleMode.FIRST);
}
 

import org.apache.qpid.proton.amqp.transport.ReceiverSettleMode; //导入依赖的package包/类
private void writeElement(Transfer transfer, int index) {
    switch (index) {
        case 0:
            getEncoder().writeUnsignedInteger(transfer.getHandle());
            break;
        case 1:
            getEncoder().writeUnsignedInteger(transfer.getDeliveryId());
            break;
        case 2:
            getEncoder().writeBinary(transfer.getDeliveryTag());
            break;
        case 3:
            getEncoder().writeUnsignedInteger(transfer.getMessageFormat());
            break;
        case 4:
            getEncoder().writeBoolean(transfer.getSettled());
            break;
        case 5:
            getEncoder().writeBoolean(transfer.getMore());
            break;
        case 6:
            ReceiverSettleMode rcvSettleMode = transfer.getRcvSettleMode();
            getEncoder().writeObject(rcvSettleMode == null ? null : rcvSettleMode.getValue());
            break;
        case 7:
            getEncoder().writeObject(transfer.getState());
            break;
        case 8:
            getEncoder().writeBoolean(transfer.getResume());
            break;
        case 9:
            getEncoder().writeBoolean(transfer.getAborted());
            break;
        case 10:
            getEncoder().writeBoolean(transfer.getBatchable());
            break;
        default:
            throw new IllegalArgumentException("Unknown Transfer value index: " + index);
    }
}
 

import org.apache.qpid.proton.amqp.transport.ReceiverSettleMode; //导入依赖的package包/类
public int size()
{
    return _attach.getProperties() != null
              ? 14
              : _attach.getDesiredCapabilities() != null
              ? 13
              : _attach.getOfferedCapabilities() != null
              ? 12
              : _attach.getMaxMessageSize() != null
              ? 11
              : _attach.getInitialDeliveryCount() != null
              ? 10
              : _attach.getIncompleteUnsettled()
              ? 9
              : _attach.getUnsettled() != null
              ? 8
              : _attach.getTarget() != null
              ? 7
              : _attach.getSource() != null
              ? 6
              : (_attach.getRcvSettleMode() != null && !_attach.getRcvSettleMode().equals(ReceiverSettleMode.FIRST))
              ? 5
              : (_attach.getSndSettleMode() != null && !_attach.getSndSettleMode().equals(SenderSettleMode.MIXED))
              ? 4
              : 3;

}
 

import org.apache.qpid.proton.amqp.transport.ReceiverSettleMode; //导入依赖的package包/类
@Override
public IncomingLink createInboundLink(String address, ReliabilityMode mode, CreditMode creditMode)
  throws MessagingException {
  checkClosed();
  String name = UUID.randomUUID().toString();
  Receiver receiver = _ssn.receiver(name);
  Source source = new Source();
  // Target target = new Target();
  if (address == null || address.isEmpty() || address.equals("#")) {
    source.setDynamic(true);
  } else {
    source.setAddress(address);
  }
  receiver.setSource(source);
  // receiver.setTarget(target);
  switch (mode) {
    case UNRELIABLE:
      receiver.setReceiverSettleMode(ReceiverSettleMode.FIRST);
      receiver.setSenderSettleMode(SenderSettleMode.SETTLED);
      break;
    case AT_LEAST_ONCE:
      receiver.setReceiverSettleMode(ReceiverSettleMode.FIRST);
      receiver.setSenderSettleMode(SenderSettleMode.UNSETTLED);
      break;
    // case EXACTLY_ONCE:
    // receiver.setReceiverSettleMode(ReceiverSettleMode.SECOND);
    // receiver.setSenderSettleMode(SenderSettleMode.UNSETTLED);
    // break;
  }
  receiver.open();

  IncomingLinkImpl inLink = new IncomingLinkImpl(this, address, receiver, mode, creditMode);
  inLink.setDynamicAddress(source.getDynamic());
  _links.put(receiver, inLink);
  receiver.setContext(inLink);
  return inLink;
}
 

import org.apache.qpid.proton.amqp.transport.ReceiverSettleMode; //导入依赖的package包/类
@Override
void onDelivery(Delivery d) {
  Link link = d.getLink();
  if (link instanceof Receiver) {
    if (d.isReadable() && !d.isPartial()) {
      Receiver receiver = (Receiver) link;
      byte[] bytes = new byte[d.pending()];
      int read = receiver.recv(bytes, 0, bytes.length);
      Message pMsg = Proton.message();
      pMsg.decode(bytes, 0, read);
      receiver.advance();

      IncomingLinkImpl inLink = (IncomingLinkImpl) link.getContext();
      SessionImpl ssn = inLink.getSession();
      InboundMessage msg = new InboundMessage(ssn.getID(), d.getTag(), ssn.getNextIncommingSequence(),
        d.isSettled(), pMsg);
      if (link.getSenderSettleMode() != SenderSettleMode.SETTLED) {
        ssn.addUnsettled(msg.getSequence(), d);
      }
      eventListener.onMessage(inLink, msg);
    } else if (d.isUpdated() && d.isSettled()) {
      if (link.getReceiverSettleMode() == ReceiverSettleMode.SECOND) {
        d.settle();
      }
    }
  } else {
    if (d.remotelySettled()) {
      TrackerImpl tracker = (TrackerImpl) d.getContext();
      tracker.setDisposition(d.getRemoteState());
      tracker.markSettled();
      eventListener.onSettled((OutgoingLinkImpl) link.getContext(), tracker);
    }
  }
}
 

import org.apache.qpid.proton.amqp.transport.ReceiverSettleMode; //导入依赖的package包/类
@Override
protected Receiver createEndpoint(JmsSessionInfo resourceInfo) {
    Receiver receiver = getParent().getEndpoint().receiver(linkName);
    receiver.setTarget(new Target());
    receiver.setSenderSettleMode(SenderSettleMode.UNSETTLED);
    receiver.setReceiverSettleMode(ReceiverSettleMode.FIRST);

    if (!hasClientID) {
      // We are trying to unsubscribe a 'global' shared subs using a 'null source lookup', add link
      // desired capabilities as hints to the peer to consider this when trying to attach the link.
      receiver.setDesiredCapabilities(new Symbol[] { AmqpSupport.SHARED, AmqpSupport.GLOBAL });
    }

    return receiver;
}
 

import org.apache.qpid.proton.amqp.transport.ReceiverSettleMode; //导入依赖的package包/类
@Override
protected Sender createEndpoint(JmsTemporaryDestination resourceInfo) {
    // Form a link name, use the local generated name with a prefix to aid debugging
    String localDestinationName = resourceInfo.getAddress();
    String senderLinkName = null;
    if (resourceInfo.isQueue()) {
        senderLinkName = "qpid-jms:" + TEMP_QUEUE_CREATOR + localDestinationName;
    } else {
        senderLinkName = "qpid-jms:" + TEMP_TOPIC_CREATOR + localDestinationName;
    }

    // Just use a bare Source, this is a producer which
    // wont send anything and the link name is unique.
    Source source = new Source();

    Target target = new Target();
    target.setDynamic(true);
    target.setDurable(TerminusDurability.NONE);
    target.setExpiryPolicy(TerminusExpiryPolicy.LINK_DETACH);

    // Set the dynamic node lifetime-policy
    Map<Symbol, Object> dynamicNodeProperties = new HashMap<Symbol, Object>();
    dynamicNodeProperties.put(DYNAMIC_NODE_LIFETIME_POLICY, DeleteOnClose.getInstance());
    target.setDynamicNodeProperties(dynamicNodeProperties);

    // Set the capability to indicate the node type being created
    if (resourceInfo.isQueue()) {
        target.setCapabilities(AmqpDestinationHelper.TEMP_QUEUE_CAPABILITY);
    } else {
        target.setCapabilities(AmqpDestinationHelper.TEMP_TOPIC_CAPABILITY);
    }

    Sender sender = getParent().getEndpoint().sender(senderLinkName);
    sender.setSource(source);
    sender.setTarget(target);
    sender.setSenderSettleMode(SenderSettleMode.UNSETTLED);
    sender.setReceiverSettleMode(ReceiverSettleMode.FIRST);

    return sender;
}