Java 类javax.jms.Message 实例源码

@Test
public void messageGetsPublished() throws JMSException, InterruptedException {
  final TestConsumer testConsumer = new TestConsumer();
  cut.subscribe(testConsumer);

  EventMessage<?> eventMessage = GenericEventMessage
          .asEventMessage("SomePayload")
          .withMetaData(MetaData.with("key", "value"));

  Message jmsMessage = converter.createJmsMessage(eventMessage, topicSession);

  publisher.publish(jmsMessage);

  Thread.sleep(1000L);

  assertNotNull(testConsumer.latest);
}

public void validateFailOnUnsupportedMessageTypeOverJNDI() throws Exception {
    final String destinationName = "testQueue";
    JmsTemplate jmsTemplate = CommonTest.buildJmsJndiTemplateForDestination(false);

    jmsTemplate.send(destinationName, new MessageCreator() {
        @Override
        public Message createMessage(Session session) throws JMSException {
            return session.createObjectMessage();
        }
    });

    JMSConsumer consumer = new JMSConsumer(jmsTemplate, mock(ComponentLog.class));
    try {
        consumer.consume(destinationName, new ConsumerCallback() {
            @Override
            public void accept(JMSResponse response) {
                // noop
            }
        });
    } finally {
        ((CachingConnectionFactory) jmsTemplate.getConnectionFactory()).destroy();
    }
}

private void addData(Message message, String text) throws JMSException {
    MessageData data = new MessageData();
    data.setPayload(text);
    Map<String, String> headers = new HashMap<>();
    Enumeration<String> names = message.getPropertyNames();
    while (names.hasMoreElements()) {
        String name = names.nextElement();
        String value = message.getStringProperty(name);
        headers.put(name, value);
    }
    data.setHeaders(headers);
    messages.add(data);
}

private void startConsumer() {
    try {
        connection = connectionFactory.createConnection();
        connection.start();
        session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        Destination destination;
        if ("topic".equalsIgnoreCase(destinationType)) {
            destination = session.createTopic(destinationName);
        } else {
            destination = session.createQueue(destinationName);
        }

        consumer = session.createConsumer(destination);
        isStarted.compareAndSet(false, true);
        while (true) {
            Message message = consumer.receive();

            if (message instanceof TextMessage) {
                TextMessage textMessage = (TextMessage) message;
                String text = textMessage.getText();
                if (isRecording.get()) {
                    addData(message, text);
                }
                counter.incrementAndGet();
            }
        }

    } catch (Exception e) {
        //e.printStackTrace();
    } finally {
        terminate();
    }
}

@Test
public void testStringBodyIsApplied() throws JMSException {
    JMSProducer producer = context.createProducer();

    final String bodyValue = "String-Value";
    final AtomicBoolean bodyValidated = new AtomicBoolean();

    MockJMSConnection connection = (MockJMSConnection) context.getConnection();
    connection.addConnectionListener(new MockJMSDefaultConnectionListener() {

        @Override
        public void onMessageSend(MockJMSSession session, Message message) throws JMSException {
            assertEquals(bodyValue, message.getBody(String.class));
            bodyValidated.set(true);
        }
    });

    producer.send(JMS_DESTINATION, bodyValue);
    assertTrue(bodyValidated.get());
}

@Test
public void testMapBodyIsApplied() throws JMSException {
    JMSProducer producer = context.createProducer();

    final Map<String, Object> bodyValue = new HashMap<String, Object>();

    bodyValue.put("Value-1", "First");
    bodyValue.put("Value-2", "Second");

    final AtomicBoolean bodyValidated = new AtomicBoolean();

    MockJMSConnection connection = (MockJMSConnection) context.getConnection();
    connection.addConnectionListener(new MockJMSDefaultConnectionListener() {

        @Override
        public void onMessageSend(MockJMSSession session, Message message) throws JMSException {
            assertEquals(bodyValue, message.getBody(Map.class));
            bodyValidated.set(true);
        }
    });

    producer.send(JMS_DESTINATION, bodyValue);
    assertTrue(bodyValidated.get());
}

/**
 * At the moment the only two supported message types are TextMessage and
 * BytesMessage which is sufficient for the type if JMS use cases NiFi is
 * used. The may change to the point where all message types are supported
 * at which point this test will no be longer required.
 */
@Test(expected = IllegalStateException.class)
public void validateFailOnUnsupportedMessageType() throws Exception {
    final String destinationName = "testQueue";
    JmsTemplate jmsTemplate = CommonTest.buildJmsTemplateForDestination(false);

    jmsTemplate.send(destinationName, new MessageCreator() {
        @Override
        public Message createMessage(Session session) throws JMSException {
            return session.createObjectMessage();
        }
    });

    JMSConsumer consumer = new JMSConsumer(jmsTemplate, mock(ComponentLog.class));
    try {
        consumer.consume(destinationName, new ConsumerCallback() {
            @Override
            public void accept(JMSResponse response) {
                // noop
            }
        });
    } finally {
        ((CachingConnectionFactory) jmsTemplate.getConnectionFactory()).destroy();
    }
}

@Test
public void testRuntimeExceptionFromSendByteBody() throws JMSException {
    JMSProducer producer = context.createProducer();

    MockJMSConnection connection = (MockJMSConnection) context.getConnection();
    connection.addConnectionListener(new MockJMSDefaultConnectionListener() {

        @Override
        public void onMessageSend(MockJMSSession session, Message message) throws JMSException {
            throw new IllegalStateException("Send Failed");
        }
    });

    try {
        producer.send(context.createTemporaryQueue(), new byte[0]);
        fail("Should have thrown an exception");
    } catch (IllegalStateRuntimeException isre) {}
}

@Test
public void testRuntimeExceptionFromSendSerializableBody() throws JMSException {
    JMSProducer producer = context.createProducer();

    MockJMSConnection connection = (MockJMSConnection) context.getConnection();
    connection.addConnectionListener(new MockJMSDefaultConnectionListener() {

        @Override
        public void onMessageSend(MockJMSSession session, Message message) throws JMSException {
            throw new IllegalStateException("Send Failed");
        }
    });

    try {
        producer.send(context.createTemporaryQueue(), UUID.randomUUID());
        fail("Should have thrown an exception");
    } catch (IllegalStateRuntimeException isre) {}
}

@Test
public void testRuntimeExceptionFromSendStringBody() throws JMSException {
    JMSProducer producer = context.createProducer();

    MockJMSConnection connection = (MockJMSConnection) context.getConnection();
    connection.addConnectionListener(new MockJMSDefaultConnectionListener() {

        @Override
        public void onMessageSend(MockJMSSession session, Message message) throws JMSException {
            throw new IllegalStateException("Send Failed");
        }
    });

    try {
        producer.send(context.createTemporaryQueue(), "test");
        fail("Should have thrown an exception");
    } catch (IllegalStateRuntimeException isre) {}
}

@Override
public void onMessage(final Message message) {
    BrpNu.set(DatumUtil.nuAlsZonedDateTime());
    try {
        LOGGER.debug("onMessage");
        final TextMessage textMessage = (TextMessage) message;
        final String text = textMessage.getText();
        final MaakSelectieResultaatTaak
                maakSelectieResultaatTaak =
                JSON_STRING_SERIALISEERDER.deserialiseerVanuitString(text, MaakSelectieResultaatTaak.class);
        final SoortSelectie soortSelectie = maakSelectieResultaatTaak.getSoortSelectie();
        if (soortSelectie == SoortSelectie.STANDAARD_SELECTIE) {
            maakSelectieResultaatTaakVerwerkerServiceImpl.verwerk(maakSelectieResultaatTaak);
        } else if (soortSelectie == SoortSelectie.SELECTIE_MET_PLAATSING_AFNEMERINDICATIE
                || soortSelectie == SoortSelectie.SELECTIE_MET_VERWIJDERING_AFNEMERINDICATIE) {
            afnemerindicatieMaakSelectieResultaatTaakVerwerkerServiceImpl.verwerk(maakSelectieResultaatTaak);
        }
    } catch (JMSException e) {
        LOGGER.error("error on message", e);
    }
}

@Override
public void onMessage(final Message msg)
{
    try
    {
        taskPool.submit(new Callable<Boolean>()
        {
            @Override
            public Boolean call() throws Exception
            {
                return processMessage(msg);
            }
        });
    }
    catch (RejectedExecutionException e)
    {
        log.error("error while submitting message task, message: {}", msg, e);
    }
}

private void doSendTextMessage( final Session session, final Destination destination,
                                final String textMessage,
                                final Map<String, ?> properties ) throws JMSException {

    try {
        final Message message = textMessage != null
                                                    ? session.createTextMessage(textMessage)
                                                    : session.createTextMessage();
        if (properties != null) {
            // Note: Setting any properties (including JMS fields) using
            // setObjectProperty might not be supported by all providers
            // Tested with: ActiveMQ
            for (final Entry<String, ?> property : properties.entrySet()) {
                message.setObjectProperty(property.getKey(), property.getValue());
            }
        }
        final MessageProducer producer = session.createProducer(destination);
        producer.send(message);
    } finally {
        releaseSession(false);
    }
}

/**
 * Since we use a request/response communication style with the client,
 * we must ensure that tha appropriate fields are set.
 */
private boolean isValidRequestResponse(Message incoming) {
    try {
        if (incoming.getJMSCorrelationID() == null) {
            getLogger().warn("JMSCorrelationID is not set! Will not process request");
            return false;
        }

        if (incoming.getJMSReplyTo() == null) {
            getLogger().warn("JMSReplyTo is not set! Will not process request");
            return false;
        }
    } catch (JMSException e) {
        getLogger().warn(
                "Failed to read JMSCorrelationID/JMSReplyTo. " +
                "Will not process request. Exception message = {}", e.getMessage());
        return false;
    }

    return true;
}

private void sendWithReplyToTemp(ConnectionFactory cf, String serviceQueue) throws JMSException, InterruptedException {
    Connection connection = cf.createConnection();
    connection.start();
    Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    TemporaryQueue tempQueue = session.createTemporaryQueue();
    TextMessage msg = session.createTextMessage("Request");
    msg.setJMSReplyTo(tempQueue);
    MessageProducer producer = session.createProducer(session.createQueue(serviceQueue));
    producer.send(msg);

    MessageConsumer consumer = session.createConsumer(tempQueue);
    Message replyMsg = consumer.receive();
    assertNotNull(replyMsg);

    LOG.debug("Reply message: {}", replyMsg);

    consumer.close();

    producer.close();
    session.close();
    connection.close();
}

/**
 * Build span and inject. Should be used by producers.
 *
 * @param message JMS message
 * @return span
 */
public static Span buildAndInjectSpan(Destination destination, final Message message,
    Tracer tracer) {
  Tracer.SpanBuilder spanBuilder = tracer.buildSpan(TracingMessageUtils.OPERATION_NAME_SEND)
      .ignoreActiveSpan()
      .withTag(Tags.SPAN_KIND.getKey(), Tags.SPAN_KIND_PRODUCER);

  SpanContext parent = TracingMessageUtils.extract(message, tracer);

  if (parent != null) {
    spanBuilder.asChildOf(parent);
  }

  Span span = spanBuilder.start();

  SpanJmsDecorator.onRequest(destination, span);

  TracingMessageUtils.inject(span, message, tracer);
  return span;
}

@Override
public void onMessage(Message msg) {
    if (msg instanceof TextMessage) {
        TextMessage textMessage = (TextMessage) msg;
        String text = "";
        try {
            text = textMessage.getText();
        } catch (JMSException e) {
            logger.error(e.getMessage());
            e.printStackTrace();
        }
        System.out.println("Received: " + text);
    } else {
        System.out.println("Received: " + msg);
    }
}

@Test
public void validateJmsHeadersAndPropertiesAreTransferredFromFFAttributesOverJNDI() throws Exception {
    final String destinationName = "testQueue";
    JmsTemplate jmsTemplate = CommonTest.buildJmsJndiTemplateForDestination(false);

    JMSPublisher publisher = new JMSPublisher(jmsTemplate, mock(ComponentLog.class));
    Map<String, String> flowFileAttributes = new HashMap<>();
    flowFileAttributes.put("foo", "foo");
    flowFileAttributes.put(JmsHeaders.REPLY_TO, "myTopic");
    publisher.publish(destinationName, "hellomq".getBytes(), flowFileAttributes);

    Message receivedMessage = jmsTemplate.receive(destinationName);
    assertTrue(receivedMessage instanceof BytesMessage);
    assertEquals("foo", receivedMessage.getStringProperty("foo"));
    assertTrue(receivedMessage.getJMSReplyTo() instanceof Topic);
    assertEquals("myTopic", ((Topic) receivedMessage.getJMSReplyTo()).getTopicName());

    ((CachingConnectionFactory) jmsTemplate.getConnectionFactory()).destroy();
}

public void purgeMessages() {
    try {
        consumer.setMessageListener(null);
        while (true) {
            Message message = consumer.receive(50);
            if (message == null) {
                logger.info("*** no more messages to purge: {}", queueName);
                break;
            }
            logger.info("*** purged message: {} - {}", queueName, message);
        }
    } catch (Exception e) {
        throw new RuntimeException(e);
    }
}

@Override
public Message createMessage() {
    try {
        return getSession().createMessage();
    } catch (JMSException jmse) {
        throw JMSExceptionSupport.createRuntimeException(jmse);
    }
}

/**
 * Extract span context from JMS message properties or active span
 *
 * @param message JMS message
 * @param tracer Tracer
 * @return extracted span context
 */
public static SpanContext extract(Message message, Tracer tracer) {
  SpanContext spanContext = tracer
      .extract(Format.Builtin.TEXT_MAP, new JmsTextMapExtractAdapter(message));
  if (spanContext != null) {
    return spanContext;
  }

  Span span = tracer.activeSpan();
  if (span != null) {
    return span.context();
  }
  return null;
}

@Override
public void emit(final Event event) {
    requireNonNull(event, "Cannot emit a null event!");

    service.serialize(event).ifPresent(json -> {
        try {
            final Message message = session.createTextMessage(json);
            message.setStringProperty("Content-Type", "application/ld+json");
            producer.send(message);
        } catch (final JMSException ex) {
            LOGGER.error("Error writing to broker: {}", ex.getMessage());
        }
    });
}

@Test
public void testProcessPartialBatch() throws Exception {
  when(messageConsumer.receiveNoWait()).thenReturn(message, (Message)null);
  source.configure(context);
  source.start();
  Assert.assertEquals(Status.READY, source.process());
  Assert.assertEquals(2, events.size());
  assertBodyIsExpected(events);
  verify(consumer).commit();
}

@Override
public void publish(final Message message, final KafkaTemplate<String, byte[]> kafkaTemplate) {

    try {
        final byte[] effectivePayload = JmsUtil.getPayload((BytesMessage) message);
        final String sourceTopic = message.getStringProperty("JMSDestination");
        final String destTopic = this.getDestTopic(sourceTopic);
        kafkaTemplate.send(destTopic, effectivePayload);
    } catch (final Throwable t) {
        LOG.error("Exception occured", t);
    }
}

@Override
public void send(Message message, int deliveryMode, int priority, long timeToLive, CompletionListener listener) throws JMSException {
    checkClosed();

    if (anonymousProducer) {
        throw new UnsupportedOperationException("Using this method is not supported on producers created without an explicit Destination");
    }

    if (listener == null) {
        throw new IllegalArgumentException("JmsCompletetionListener cannot be null");
    }

    sendMessage(destination, message, deliveryMode, priority, timeToLive, listener);
}

@Override
public void onMessage(Message message) {
    try {
        if (message instanceof TextMessage) {
            logger.debug("got message: " + message.getJMSCorrelationID());
            TextMessage textMessage = (TextMessage) message;
            msgProcessor.processMessage(textMessage.getText(), getMessageId(textMessage), getMessageHeaders(textMessage));
        }
    } catch (JMSException e) {
        logger.error("JMSException in onMessage", e);
    }
}

/**
 *
 *
 */
private Map<String, Object> extractMessageHeaders(Message message) {
    // even though all values are Strings in current impl, it may change in
    // the future, so keeping it <String, Object>
    Map<String, Object> messageHeaders = new HashMap<>();
    try {
        messageHeaders.put(JmsHeaders.DELIVERY_MODE, String.valueOf(message.getJMSDeliveryMode()));
        messageHeaders.put(JmsHeaders.EXPIRATION, String.valueOf(message.getJMSExpiration()));
        messageHeaders.put(JmsHeaders.PRIORITY, String.valueOf(message.getJMSPriority()));
        messageHeaders.put(JmsHeaders.REDELIVERED, String.valueOf(message.getJMSRedelivered()));
        messageHeaders.put(JmsHeaders.TIMESTAMP, String.valueOf(message.getJMSTimestamp()));
        messageHeaders.put(JmsHeaders.CORRELATION_ID, message.getJMSCorrelationID());
        messageHeaders.put(JmsHeaders.MESSAGE_ID, message.getJMSMessageID());
        messageHeaders.put(JmsHeaders.TYPE, message.getJMSType());

        String replyToDestinationName = this.retrieveDestinationName(message.getJMSReplyTo(), JmsHeaders.REPLY_TO);
        if (replyToDestinationName != null) {
            messageHeaders.put(JmsHeaders.REPLY_TO, replyToDestinationName);
        }
        String destinationName = this.retrieveDestinationName(message.getJMSDestination(), JmsHeaders.DESTINATION);
        if (destinationName != null) {
            messageHeaders.put(JmsHeaders.DESTINATION, destinationName);
        }
    } catch (Exception e) {
        throw new IllegalStateException("Failed to extract JMS Headers", e);
    }
    return messageHeaders;
}

@Override
public void onMessage(Message message) {
    try {
        if (message instanceof TextMessage) {
            logger.debug("got message: " + message.getJMSCorrelationID());
            TextMessage textMessage = (TextMessage) message;
            msgProcessor.processMessage(textMessage.getText(), getMessageType(textMessage), getMessageId(textMessage));
        }
    } catch (JMSException e) {
        logger.error("JMSException in onMessage", e);
    }
}

@Override
protected Message toMessage(Object object, Session session, ObjectWriter objectWriter)
        throws JMSException, MessageConversionException {
    Message message = super.toMessage(object, session, objectWriter);
    this.setDuplicateDetectionId(object, message);
    return message;
}

@Override
public void publiceerSelectieTaakResultaat(final SelectieTaakResultaat selectieTaakResultaat) {
    final ProducerCallback<Void> producerCallback = (final Session session, final MessageProducer producer) -> {
        final Message message = session.createTextMessage(serializer.serialiseerNaarString(selectieTaakResultaat));
        producer.send(message);
        return null;
    };
    publiceer(selectieTaakResultaatJmsTemplate, producerCallback, () -> "fout in verzenden berichten naar selectie taak resultaat queue");
}

/**
 * Browse messages using Queue Browser.
 * By default, you browse all actual messages in the queue.
 * Messages may be arriving and expiring while the scan is done.
 */
void browseMessages() throws Exception {
    Connection conn = createConnection(clientOptions);
    Session ssn = createSession(clientOptions, conn, transacted);
    QueueBrowser qBrowser = ssn.createBrowser((Queue) getDestination(), msgSelector);
    conn.start();
    Enumeration<?> enumMsgs = qBrowser.getEnumeration();
    while (enumMsgs.hasMoreElements()) {
        Message msg = (Message) enumMsgs.nextElement();
        printMessage(clientOptions, msg);
    }
    close(conn);
}

/**
 * {@inheritDoc}
 */
@Override
public Message receive(long timeout) throws JMSException {
    Message message = delegate.receive(timeout);
    if (!(delegate.getMessageListener() instanceof PreservesMessagePropertiesMessageListener)) {
        copyFromMessage(message);
    }
    return message;
}

public Message getMessage(MessageContext messageContext,
        ProxyMessageConsumer proxyMessageConsumer) throws JMSException {
    String destinationName = proxyMessageConsumer.getDestination()
            .toString();
    Destination destination = destinationMap.get(destinationName);

    if (destination instanceof Topic) {
        return messageHandler.consumeMessageFromTopic(messageContext,
                destinationName, proxyMessageConsumer.getId());
    } else {
        return messageHandler.consumeMessageFromQueue(messageContext,
                destinationName);
    }
}

@Override
public void sendMessage(MessageDto payload) throws JMSException {

  Message message = MessageConvertUtils.toMessage(messageConverter, session, payload);
  dupMessageDetectStrategy.setId(message, payload);
  messageProducer.send(message);

}

public void sendMessage(Message message) throws JMSException {
    Connection connection = null;
    try {
        connection = startConnection(); //try to be smarter here and initiate start connection
        Session session = null;
        try {
            session = connection.createSession(isTransacted, Session.AUTO_ACKNOWLEDGE);
            Destination dest;
            if (isQueue) {
                dest = session.createQueue(destinationName);
            } else {
                dest = session.createTopic(destinationName);
            }
            MessageProducer producer = session.createProducer(dest);
            try {

                if (isPersistant) producer.setDeliveryMode(DeliveryMode.PERSISTENT);
                if (timeToLive > 0) producer.setTimeToLive(timeToLive);

                producer.send(message);
            } finally {
                if (producer != null) producer.close();
            }
        } finally {
            if (session != null) session.close();
        }
    } finally {
        safeCloseConnection(connection);
    }
}

@Override
public void onMessage(Message message) {
  if (message instanceof TextMessage) {
    TextMessage textMessage = (TextMessage) message;
    processMessage(textMessage);
  }
}

public void send() {
    template.send(this.destination, new MessageCreator() {
        @Override
        public Message createMessage(Session session) throws JMSException {
            TextMessage message=session.createTextMessage();
            System.out.println("发出消息,延时10秒.");
            String msgId=message.getJMSMessageID();
            System.out.println("msgId:"+msgId);
            message.setLongProperty(ScheduledMessage.AMQ_SCHEDULED_DELAY, 20*1000);
            message.setStringProperty("content", "内容content...");
            return message;
        }
    });
}

@Parameters({ "broker-ssl-port"})
@Test
public void testConsumerProducerWithSsl(String port) throws Exception {
    String queueName = "testConsumerProducerWithAutoAck";
    InitialContext initialContextForQueue = ClientHelper
            .getInitialContextBuilder("admin", "admin", "localhost", port)
            .enableSsl()
            .withQueue(queueName)
            .build();

    ConnectionFactory connectionFactory
            = (ConnectionFactory) initialContextForQueue.lookup(ClientHelper.CONNECTION_FACTORY);
    Connection connection = connectionFactory.createConnection();
    connection.start();

    // publish 100 messages
    Session producerSession = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    Queue queue = producerSession.createQueue(queueName);
    MessageProducer producer = producerSession.createProducer(queue);

    int numberOfMessages = 100;
    for (int i = 0; i < numberOfMessages; i++) {
        producer.send(producerSession.createTextMessage("Test message " + i));
    }
    producerSession.close();

    // Consume published messages
    Session subscriberSession = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    Destination subscriberDestination = (Destination) initialContextForQueue.lookup(queueName);
    MessageConsumer consumer = subscriberSession.createConsumer(subscriberDestination);

    for (int i = 0; i < numberOfMessages; i++) {
        Message message = consumer.receive(1000);
        Assert.assertNotNull(message, "Message #" + i + " was not received");
    }

    connection.close();
}

@Override
public Message receiveNoWait() {
    try {
        return context.setLastMessage(consumer.receiveNoWait());
    } catch (JMSException e) {
        throw Utils.convertToRuntimeException(e);
    }
}

/**
 * Receive a message from queue for a period of time
 *
 * @param queueName queue name
 * @param timeout timeout period in milliseconds
 * @return the received message
 */
@PublicAtsApi
public Message receiveMessageFromQueue( final String queueName, final long timeout ) {

    try {
        final Session session = loadSession(false, Session.AUTO_ACKNOWLEDGE);
        return doReceiveMessage(session, session.createQueue(queueName), timeout);
    } catch (Exception e) {
        throw new JmsMessageException("Failed to read message message from queue " + queueName, e);
    }
}