import org.apache.qpid.proton.amqp.DescribedType; //导入依赖的package包/类
@Override
public boolean equals(Object o)
{
    if (this == o)
    {
        return true;
    }
    if (o == null || ! (o instanceof DescribedType))
    {
        return false;
    }

    DescribedType that = (DescribedType) o;

    if (_described != null ? !_described.equals(that.getDescribed()) : that.getDescribed() != null)
    {
        return false;
    }
    if (_descriptor != null ? !_descriptor.equals(that.getDescriptor()) : that.getDescriptor() != null)
    {
        return false;
    }

    return true;
}
 

import org.apache.qpid.proton.amqp.DescribedType; //导入依赖的package包/类
private void doTestDecodeUnknownDescribedTypeSeries(int size) throws IOException {
    for (int i = 0; i < size; ++i) {
        encoder.writeObject(NoLocalType.NO_LOCAL);
    }

    buffer.clear();

    for (int i = 0; i < size; ++i) {
        final Object result = decoder.readObject();

        assertNotNull(result);
        assertTrue(result instanceof DescribedType);

        DescribedType resultTye = (DescribedType) result;
        assertEquals(NoLocalType.NO_LOCAL.getDescriptor(), resultTye.getDescriptor());
    }
}
 

import org.apache.qpid.proton.amqp.DescribedType; //导入依赖的package包/类
/**
 * Search for a particular filter using a set of known indentification values
 * in the Map of filters.
 *
 * @param filters   The filters map that should be searched.
 * @param filterIds The aliases for the target filter to be located.
 * @return the filter if found in the mapping or null if not found.
 */
public static Map.Entry<Symbol, DescribedType> findFilter(Map<Symbol, Object> filters, Object[] filterIds) {

   if (filterIds == null || filterIds.length == 0) {
      throw new IllegalArgumentException("Invalid empty Filter Ids array passed: ");
   }

   if (filters == null || filters.isEmpty()) {
      return null;
   }

   for (Map.Entry<Symbol, Object> filter : filters.entrySet()) {
      if (filter.getValue() instanceof DescribedType) {
         DescribedType describedType = ((DescribedType) filter.getValue());
         Object descriptor = describedType.getDescriptor();

         for (Object filterId : filterIds) {
            if (descriptor.equals(filterId)) {
               return new AbstractMap.SimpleImmutableEntry<>(filter.getKey(), describedType);
            }
         }
      }
   }

   return null;
}
 

import org.apache.qpid.proton.amqp.DescribedType; //导入依赖的package包/类
void receiveFrame(int type, int channel, DescribedType describedType, Binary payload)
{
    Handler handler = getFirstHandler();
    if(handler == null)
    {
        Object actualDescriptor = describedType.getDescriptor();
        Object mappedDescriptor = FrameDescriptorMapping.lookupMapping(actualDescriptor);

        throw new IllegalStateException("No handler! Received frame, descriptor=" + actualDescriptor + "/" + mappedDescriptor);
    }

    if(handler instanceof FrameHandler)
    {
        ((FrameHandler)handler).frame(type, channel, describedType, payload, this);
        removeFirstHandler();
    }
    else
    {
        throw new IllegalStateException("Received frame but the next handler is a " + handler);
    }
}
 

import org.apache.qpid.proton.amqp.DescribedType; //导入依赖的package包/类
public void expectLinkFlowRespondWithTransfer(final HeaderDescribedType headerDescribedType,
                                              final MessageAnnotationsDescribedType messageAnnotationsDescribedType,
                                              final PropertiesDescribedType propertiesDescribedType,
                                              ApplicationPropertiesDescribedType appPropertiesDescribedType,
                                              final DescribedType content,
                                              final int count,
                                              final boolean drain,
                                              final boolean sendDrainFlowResponse,
                                              Matcher<UnsignedInteger> creditMatcher,
                                              final Integer nextIncomingId,
                                              boolean addMessageNumberProperty)
{
    expectLinkFlowRespondWithTransfer(headerDescribedType, messageAnnotationsDescribedType, propertiesDescribedType,
                                      appPropertiesDescribedType, content, count, drain, sendDrainFlowResponse,
                                      creditMatcher, nextIncomingId, false, addMessageNumberProperty);
}
 

import org.apache.qpid.proton.amqp.DescribedType; //导入依赖的package包/类
/**
 * @param receivedBinary
 *      The received Binary value that should be validated.
 *
 * @return the number of bytes consumed from the provided Binary
 *
 * @throws RuntimeException if the provided Binary does not match expectation in some way
 */
public int verify(Binary receivedBinary) throws RuntimeException
{
    int length = receivedBinary.getLength();
    Data data = Data.Factory.create();
    long decoded = data.decode(receivedBinary.asByteBuffer());
    if(decoded > Integer.MAX_VALUE)
    {
        throw new IllegalStateException("Decoded more bytes than Binary supports holding");
    }

    if(decoded < length && !_expectTrailingBytes)
    {
        throw new IllegalArgumentException("Expected to consume all bytes, but trailing bytes remain: Got "
                                    + length + ", consumed "+ decoded);
    }

    DescribedType decodedDescribedType = data.getDescribedType();
    verifyReceivedDescribedType(decodedDescribedType);

    //Need to cast to int, but verified earlier that it is < Integer.MAX_VALUE
    return (int) decoded;
}
 

import org.apache.qpid.proton.amqp.DescribedType; //导入依赖的package包/类
protected void configureSource(Source source) {
    Map<Symbol, DescribedType> filters = new HashMap<Symbol, DescribedType>();

    if (info.getSubscriptionName() != null && !info.getSubscriptionName().isEmpty()) {
        source.setExpiryPolicy(TerminusExpiryPolicy.NEVER);
        source.setDurable(TerminusDurability.UNSETTLED_STATE);
        source.setDistributionMode(COPY);
    } else {
        source.setDurable(TerminusDurability.NONE);
        source.setExpiryPolicy(TerminusExpiryPolicy.LINK_DETACH);
    }

    if (info.isNoLocal()) {
        filters.put(JMS_NO_LOCAL_SYMBOL, AmqpJmsNoLocalType.NO_LOCAL);
    }

    if (info.getSelector() != null && !info.getSelector().trim().equals("")) {
        filters.put(JMS_SELECTOR_SYMBOL, new AmqpJmsSelectorType(info.getSelector()));
    }

    if (!filters.isEmpty()) {
        source.setFilter(filters);
    }
}
 

import org.apache.qpid.proton.amqp.DescribedType; //导入依赖的package包/类
@Override
public boolean equals(Object obj)
{

    return obj instanceof DescribedType
           && _descriptor == null ? ((DescribedType) obj).getDescriptor() == null
                                 : _descriptor.equals(((DescribedType) obj).getDescriptor())
           && _described == null ?  ((DescribedType) obj).getDescribed() == null
                                 : _described.equals(((DescribedType) obj).getDescribed());

}
 

import org.apache.qpid.proton.amqp.DescribedType; //导入依赖的package包/类
public void writeDescribedType(final DescribedType d)
{
    if(d == null)
    {
        writeNull();
    }
    else
    {
        _buffer.put(DESCRIBED_TYPE_OP);
        writeObject(d.getDescriptor());
        writeObject(d.getDescribed());
    }
}
 

import org.apache.qpid.proton.amqp.DescribedType; //导入依赖的package包/类
public void writeObject(final Object o)
{
    if (o == null)
    {
        getBuffer().put(EncodingCodes.NULL);
        return;
    }

    AMQPType type = _typeRegistry.get(o.getClass());

    if(type == null)
    {
        if(o.getClass().isArray())
        {
            writeArrayType(o);
        }
        else if(o instanceof List)
        {
            writeList((List)o);
        }
        else if(o instanceof Map)
        {
            writeMap((Map)o);
        }
        else if(o instanceof DescribedType)
        {
            writeDescribedType((DescribedType)o);
        }
        else
        {
            throw new IllegalArgumentException(
                "Do not know how to write Objects of class " + o.getClass().getName());
        }
    }
    else
    {
        type.write(o);
    }
}
 

import org.apache.qpid.proton.amqp.DescribedType; //导入依赖的package包/类
@Override
public DescribedType getValue()
{
    final Object descriptor = _first == null ? null : _first.getValue();
    Element second = _first == null ? null : _first.next();
    final Object described = second == null ? null : second.getValue();
    return new DescribedTypeImpl(descriptor,described);
}
 

import org.apache.qpid.proton.amqp.DescribedType; //导入依赖的package包/类
public TypeEncoding<DescribedType> getEncoding(final DescribedType val)
{
    TypeEncoding underlyingEncoding = _encoder.getType(val.getDescribed()).getEncoding(val.getDescribed());
    TypeEncoding encoding = _encodings.get(underlyingEncoding);
    if(encoding == null)
    {
        encoding = new DynamicDescribedTypeEncoding(underlyingEncoding);
        _encodings.put(underlyingEncoding, encoding);
    }

    return encoding;
}
 

import org.apache.qpid.proton.amqp.DescribedType; //导入依赖的package包/类
@Test
public void testDecodeUnknownDescribedType() throws Exception {
    encoder.writeObject(NoLocalType.NO_LOCAL);

    buffer.clear();

    Object result = decoder.readObject();
    assertTrue(result instanceof DescribedType);
    DescribedType resultTye = (DescribedType) result;
    assertEquals(NoLocalType.NO_LOCAL.getDescriptor(), resultTye.getDescriptor());
}
 

import org.apache.qpid.proton.amqp.DescribedType; //导入依赖的package包/类
@Test
public void testDescribedArray() throws IOException
{
    Decoder d = createDecoder(getBytes("described_array"));
    DescribedType a[] = (DescribedType[]) (d.readArray());
    for (int i = 0; i < 10; ++i)
    {
        assertEquals(Symbol.valueOf("int-array"), a[i].getDescriptor());
        assertEquals(i, a[i].getDescribed());
    }
}
 

import org.apache.qpid.proton.amqp.DescribedType; //导入依赖的package包/类
protected void configureSource(Source source) {
   Map<Symbol, DescribedType> filters = new HashMap<>();
   Symbol[] outcomes = new Symbol[] {Accepted.DESCRIPTOR_SYMBOL, Rejected.DESCRIPTOR_SYMBOL, Released.DESCRIPTOR_SYMBOL, Modified.DESCRIPTOR_SYMBOL};

   if (getSubscriptionName() != null && !getSubscriptionName().isEmpty()) {
      source.setExpiryPolicy(TerminusExpiryPolicy.NEVER);
      source.setDurable(TerminusDurability.UNSETTLED_STATE);
      source.setDistributionMode(COPY);
   } else {
      source.setDurable(TerminusDurability.NONE);
      source.setExpiryPolicy(TerminusExpiryPolicy.LINK_DETACH);
   }

   source.setOutcomes(outcomes);

   Modified modified = new Modified();
   modified.setDeliveryFailed(true);
   modified.setUndeliverableHere(false);

   source.setDefaultOutcome(modified);

   if (isNoLocal()) {
      filters.put(NO_LOCAL_NAME, AmqpNoLocalFilter.NO_LOCAL);
   }

   if (getSelector() != null && !getSelector().trim().equals("")) {
      filters.put(JMS_SELECTOR_NAME, new AmqpJmsSelectorFilter(getSelector()));
   }

   if (!filters.isEmpty()) {
      source.setFilter(filters);
   }
}
 

import org.apache.qpid.proton.amqp.DescribedType; //导入依赖的package包/类
@Test(timeout = 20000)
public void testReceiveMessageWithReceiveZeroTimeout() throws Exception {
    try (TestAmqpPeer testPeer = new TestAmqpPeer();) {
        JMSContext context = testFixture.createJMSContext(testPeer);

        testPeer.expectBegin();

        Queue queue = context.createQueue("myQueue");

        DescribedType amqpValueNullContent = new AmqpValueDescribedType(null);

        testPeer.expectReceiverAttach();
        testPeer.expectLinkFlowRespondWithTransfer(null, null, null, null, amqpValueNullContent);
        testPeer.expectDispositionThatIsAcceptedAndSettled();

        JMSConsumer messageConsumer = context.createConsumer(queue);
        Message receivedMessage = messageConsumer.receive(0);

        assertNotNull("A message should have been recieved", receivedMessage);

        testPeer.expectEnd();
        testPeer.expectClose();
        context.close();

        testPeer.waitForAllHandlersToComplete(2000);
    }
}
 

import org.apache.qpid.proton.amqp.DescribedType; //导入依赖的package包/类
@Test(timeout = 20000)
public void testReceiveBodyTextMessage() throws Exception {
    try (TestAmqpPeer testPeer = new TestAmqpPeer();) {
        JMSContext context = testFixture.createJMSContext(testPeer);

        testPeer.expectBegin();

        final String content = "Message-Content";
        Queue queue = context.createQueue("myQueue");

        DescribedType amqpValueContent = new AmqpValueDescribedType(content);

        testPeer.expectReceiverAttach();
        testPeer.expectLinkFlowRespondWithTransfer(null, null, null, null, amqpValueContent);
        testPeer.expectDispositionThatIsAcceptedAndSettled();
        testPeer.expectEnd();
        testPeer.expectClose();

        JMSConsumer messageConsumer = context.createConsumer(queue);
        String received = messageConsumer.receiveBody(String.class, 3000);

        assertNotNull(received);
        assertEquals(content, received);

        context.close();

        testPeer.waitForAllHandlersToComplete(3000);
    }
}
 

import org.apache.qpid.proton.amqp.DescribedType; //导入依赖的package包/类
@Test(timeout = 20000)
public void testReceiveBodyBytesMessage() throws Exception {
    try (TestAmqpPeer testPeer = new TestAmqpPeer();) {
        JMSContext context = testFixture.createJMSContext(testPeer);

        testPeer.expectBegin();

        Queue queue = context.createQueue("myQueue");

        PropertiesDescribedType properties = new PropertiesDescribedType();
        properties.setContentType(Symbol.valueOf(AmqpMessageSupport.OCTET_STREAM_CONTENT_TYPE));

        MessageAnnotationsDescribedType msgAnnotations = null;
        msgAnnotations = new MessageAnnotationsDescribedType();
        msgAnnotations.setSymbolKeyedAnnotation(AmqpMessageSupport.JMS_MSG_TYPE, AmqpMessageSupport.JMS_BYTES_MESSAGE);

        final byte[] expectedContent = "expectedContent".getBytes();
        DescribedType dataContent = new DataDescribedType(new Binary(expectedContent));

        testPeer.expectReceiverAttach();
        testPeer.expectLinkFlowRespondWithTransfer(null, msgAnnotations, properties, null, dataContent);
        testPeer.expectDispositionThatIsAcceptedAndSettled();

        JMSConsumer messageConsumer = context.createConsumer(queue);
        byte[] received = messageConsumer.receiveBody(byte[].class, 3000);
        testPeer.waitForAllHandlersToComplete(3000);

        assertNotNull(received);
        assertTrue(Arrays.equals(expectedContent, received));

        testPeer.expectEnd();
        testPeer.expectClose();

        context.close();

        testPeer.waitForAllHandlersToComplete(3000);
    }
}
 

import org.apache.qpid.proton.amqp.DescribedType; //导入依赖的package包/类
public void doTestReceiveBodyFailsDoesNotAcceptMessage(int sessionMode) throws Exception {
    try (TestAmqpPeer testPeer = new TestAmqpPeer();) {
        JMSContext context = testFixture.createJMSContext(testPeer);

        testPeer.expectBegin();

        final String content = "Message-Content";
        Queue queue = context.createQueue("myQueue");

        DescribedType amqpValueContent = new AmqpValueDescribedType(content);

        testPeer.expectReceiverAttach();
        testPeer.expectLinkFlowRespondWithTransfer(null, null, null, null, amqpValueContent);
        testPeer.expectEnd();
        testPeer.expectClose();

        JMSConsumer messageConsumer = context.createConsumer(queue);
        try {
            messageConsumer.receiveBody(Boolean.class, 3000);
            fail("Should not read as Boolean type");
        } catch (MessageFormatRuntimeException mfre) {
        }

        context.close();

        testPeer.waitForAllHandlersToComplete(3000);
    }
}
 

import org.apache.qpid.proton.amqp.DescribedType; //导入依赖的package包/类
@Test(timeout = 20000)
public void testReceiveMessageAndGetBody() throws Exception {
    try (TestAmqpPeer testPeer = new TestAmqpPeer();) {
        Connection connection = testFixture.establishConnecton(testPeer);
        connection.start();

        testPeer.expectBegin();

        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        Queue queue = session.createQueue("myQueue");

        DescribedType amqpValueNullContent = new AmqpValueDescribedType(null);

        testPeer.expectReceiverAttach();
        testPeer.expectLinkFlowRespondWithTransfer(null, null, null, null, amqpValueNullContent);
        testPeer.expectDispositionThatIsAcceptedAndSettled();
        testPeer.expectClose();

        MessageConsumer messageConsumer = session.createConsumer(queue);
        Message receivedMessage = messageConsumer.receive(3000);

        assertTrue(receivedMessage.isBodyAssignableTo(Object.class));
        assertTrue(receivedMessage.isBodyAssignableTo(String.class));
        assertTrue(receivedMessage.isBodyAssignableTo(byte[].class));
        assertTrue(receivedMessage.isBodyAssignableTo(Serializable.class));
        assertTrue(receivedMessage.isBodyAssignableTo(Map.class));

        assertNull(receivedMessage.getBody(Object.class));
        assertNull(receivedMessage.getBody(String.class));
        assertNull(receivedMessage.getBody(byte[].class));
        assertNull(receivedMessage.getBody(Serializable.class));
        assertNull(receivedMessage.getBody(Map.class));

        connection.close();

        testPeer.waitForAllHandlersToComplete(3000);
    }
}