javax.jms.JMSConsumer#receiveBody ( )源码实例Demo

@Test
public void testJMSContextConsumerThrowsMessageFormatExceptionOnMalformedBody() throws Exception {
   Queue queue = createQueue(true, "ContextMalformedBodyTestQueue");

   JMSContext context = qraConnectionFactory.createContext();
   JMSProducer producer = context.createProducer();

   TextMessage message = context.createTextMessage("TestMessage");
   producer.send(queue, message);

   JMSConsumer consumer = context.createConsumer(queue);

   try {
      consumer.receiveBody(Boolean.class);
      fail("Should thrown MessageFormatException");
   } catch (MessageFormatRuntimeException mfre) {
      // Do nothing test passed
   } catch (Exception e) {
      fail("Threw wrong exception, should be MessageFormatRuntimeException, instead got: " + e.getClass().getCanonicalName());
   }
}
 

@Test
public void sharedDurableSubSimpleRoundRobin() throws Exception {
   context = cf.createContext();
   try {
      JMSConsumer con1 = context.createSharedDurableConsumer(topic1, "mySharedCon");
      JMSConsumer con2 = context.createSharedDurableConsumer(topic1, "mySharedCon");
      context.start();
      JMSProducer producer = context.createProducer();
      int numMessages = 10;
      for (int i = 0; i < numMessages; i++) {
         producer.send(topic1, "msg:" + i);
      }

      for (int i = 0; i < numMessages; i += 2) {
         String msg = con1.receiveBody(String.class, 5000);
         msg = con2.receiveBody(String.class, 5000);
      }

   } finally {
      context.close();
   }
}
 

@Test
public void testReceiveBody() throws JMSException {
    JMSConsumer consumer = context.createConsumer(context.createTemporaryQueue());

    try {
        consumer.receiveBody(String.class);
        fail("Should not be able to interact with closed consumer");
    } catch (JMSRuntimeException ise) {}
}
 

@Test
public void testReceiveBodyTimed() throws JMSException {
    JMSConsumer consumer = context.createConsumer(context.createTemporaryQueue());

    try {
        consumer.receiveBody(String.class, 1);
        fail("Should not be able to interact with closed consumer");
    } catch (JMSRuntimeException ise) {}
}
 

@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);
    }
}
 

@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(AmqpMessageSupport.OCTET_STREAM_CONTENT_TYPE);

        MessageAnnotationsDescribedType msgAnnotations = null;
        msgAnnotations = new MessageAnnotationsDescribedType();
        msgAnnotations.setSymbolKeyedAnnotation(AmqpMessageSupport.JMS_MSG_TYPE.toString(), 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);
    }
}
 

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);
    }
}
 

@Test
public void testShareDuraleWithJMSContext() throws Exception {
   ((ActiveMQConnectionFactory) cf).setConsumerWindowSize(0);
   JMSContext conn = cf.createContext(JMSContext.AUTO_ACKNOWLEDGE);

   JMSConsumer consumer = conn.createSharedDurableConsumer(topic, "c1");

   JMSProducer producer = conn.createProducer();

   for (int i = 0; i < 100; i++) {
      producer.setProperty("count", i).send(topic, "test" + i);
   }

   JMSContext conn2 = conn.createContext(JMSContext.AUTO_ACKNOWLEDGE);
   JMSConsumer consumer2 = conn2.createSharedDurableConsumer(topic, "c1");

   for (int i = 0; i < 50; i++) {
      String txt = consumer.receiveBody(String.class, 5000);
      Assert.assertNotNull(txt);

      txt = consumer.receiveBody(String.class, 5000);
      Assert.assertNotNull(txt);
   }

   Assert.assertNull(consumer.receiveNoWait());
   Assert.assertNull(consumer2.receiveNoWait());

   boolean exceptionHappened = false;

   try {
      conn.unsubscribe("c1");
   } catch (Exception e) {
      e.printStackTrace();
      exceptionHappened = true;
   }

   Assert.assertTrue(exceptionHappened);

   consumer.close();
   consumer2.close();
   conn2.close();

   conn.unsubscribe("c1");

}
 

@Test(timeout = 20000)
public void testReceiveBodyMapMessage() throws Exception {
    try (TestAmqpPeer testPeer = new TestAmqpPeer();) {
        JMSContext context = testFixture.createJMSContext(testPeer);

        testPeer.expectBegin();

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

        // Prepare an AMQP message for the test peer to send, containing an
        // AmqpValue section holding a map with entries for each supported type,
        // and annotated as a JMS map message.
        String myBoolKey = "myBool";
        boolean myBool = true;
        String myByteKey = "myByte";
        byte myByte = 4;
        String myBytesKey = "myBytes";
        byte[] myBytes = myBytesKey.getBytes();
        String myCharKey = "myChar";
        char myChar = 'd';
        String myDoubleKey = "myDouble";
        double myDouble = 1234567890123456789.1234;
        String myFloatKey = "myFloat";
        float myFloat = 1.1F;
        String myIntKey = "myInt";
        int myInt = Integer.MAX_VALUE;
        String myLongKey = "myLong";
        long myLong = Long.MAX_VALUE;
        String myShortKey = "myShort";
        short myShort = 25;
        String myStringKey = "myString";
        String myString = myStringKey;

        Map<String, Object> map = new LinkedHashMap<String, Object>();
        map.put(myBoolKey, myBool);
        map.put(myByteKey, myByte);
        map.put(myBytesKey, new Binary(myBytes));// the underlying AMQP message uses Binary rather than byte[] directly.
        map.put(myCharKey, myChar);
        map.put(myDoubleKey, myDouble);
        map.put(myFloatKey, myFloat);
        map.put(myIntKey, myInt);
        map.put(myLongKey, myLong);
        map.put(myShortKey, myShort);
        map.put(myStringKey, myString);

        MessageAnnotationsDescribedType msgAnnotations = new MessageAnnotationsDescribedType();
        msgAnnotations.setSymbolKeyedAnnotation(AmqpMessageSupport.JMS_MSG_TYPE.toString(), AmqpMessageSupport.JMS_MAP_MESSAGE);

        DescribedType amqpValueSectionContent = new AmqpValueDescribedType(map);

        // receive the message from the test peer
        testPeer.expectReceiverAttach();
        testPeer.expectLinkFlowRespondWithTransfer(null, msgAnnotations, null, null, amqpValueSectionContent);
        testPeer.expectDispositionThatIsAcceptedAndSettled();
        testPeer.expectEnd();
        testPeer.expectClose();

        JMSConsumer messageConsumer = context.createConsumer(queue);
        @SuppressWarnings("unchecked")
        Map<String, Object> receivedMap = messageConsumer.receiveBody(Map.class, 3000);

        // verify the content is as expected
        assertNotNull("Map was not received", receivedMap);

        assertEquals("Unexpected boolean value", myBool, receivedMap.get(myBoolKey));
        assertEquals("Unexpected byte value", myByte, receivedMap.get(myByteKey));
        byte[] readBytes = (byte[]) receivedMap.get(myBytesKey);
        assertTrue("Read bytes were not as expected: " + Arrays.toString(readBytes), Arrays.equals(myBytes, readBytes));
        assertEquals("Unexpected char value", myChar, receivedMap.get(myCharKey));
        assertEquals("Unexpected double value", myDouble, (double) receivedMap.get(myDoubleKey), 0.0);
        assertEquals("Unexpected float value", myFloat, (float) receivedMap.get(myFloatKey), 0.0);
        assertEquals("Unexpected int value", myInt, receivedMap.get(myIntKey));
        assertEquals("Unexpected long value", myLong, receivedMap.get(myLongKey));
        assertEquals("Unexpected short value", myShort, receivedMap.get(myShortKey));
        assertEquals("Unexpected UTF value", myString, receivedMap.get(myStringKey));

        context.close();

        testPeer.waitForAllHandlersToComplete(3000);
    }
}
 

@Test(timeout = 20000)
public void testReceiveBodyObjectMessage() 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(AmqpMessageSupport.SERIALIZED_JAVA_OBJECT_CONTENT_TYPE);

        String expectedContent = "expectedContent";

        ByteArrayOutputStream baos = new ByteArrayOutputStream();
        ObjectOutputStream oos = new ObjectOutputStream(baos);
        oos.writeObject(expectedContent);
        oos.flush();
        oos.close();
        byte[] bytes = baos.toByteArray();

        MessageAnnotationsDescribedType msgAnnotations = new MessageAnnotationsDescribedType();
        msgAnnotations.setSymbolKeyedAnnotation(AmqpMessageSupport.JMS_MSG_TYPE.toString(), AmqpMessageSupport.JMS_OBJECT_MESSAGE);

        DescribedType dataContent = new DataDescribedType(new Binary(bytes));

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

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

        assertNotNull(received);
        assertEquals(expectedContent, received);

        context.close();

        testPeer.waitForAllHandlersToComplete(3000);
    }
}
 

public void doTestReceiveBodyFailsThenCalledWithCorrectType(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);

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

        testPeer.waitForAllHandlersToComplete(3000);

        if (sessionMode == JMSContext.AUTO_ACKNOWLEDGE ||
            sessionMode == JMSContext.DUPS_OK_ACKNOWLEDGE) {

            testPeer.expectDispositionThatIsAcceptedAndSettled();
        }

        String received = messageConsumer.receiveBody(String.class, 3000);

        if (sessionMode == JMSContext.AUTO_ACKNOWLEDGE ||
            sessionMode == JMSContext.DUPS_OK_ACKNOWLEDGE) {

            assertNotNull(received);
            assertEquals(content, received);
        } else {
            assertNull(received);
        }

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

        context.close();

        testPeer.waitForAllHandlersToComplete(3000);
    }
}