Java 类javax.jms.Destination 实例源码

private void putTopic(List<String> events) throws Exception {
  ConnectionFactory factory = new ActiveMQConnectionFactory(USERNAME,
      PASSWORD, BROKER_BIND_URL);
  Connection connection = factory.createConnection();
  connection.start();

  Session session = connection.createSession(true,
      Session.AUTO_ACKNOWLEDGE);
  Destination destination = session.createTopic(DESTINATION_NAME);
  MessageProducer producer = session.createProducer(destination);

  for (String event : events) {
    TextMessage message = session.createTextMessage();
    message.setText(event);
    producer.send(message);
  }
  session.commit();
  session.close();
  connection.close();
}

/**
 * 使用jmsTemplate的send/MessageCreator()发送Map类型的消息并在Message中附加属性用于消息过滤.
 */
private void sendMessage(final User user, Destination destination) {
    jmsTemplate.send(destination, new MessageCreator() {
        @Override
        public Message createMessage(Session session) throws JMSException {

            MapMessage message = session.createMapMessage();
            message.setString("userName", user.getName());
            message.setString("email", user.getEmail());

            message.setStringProperty("objectType", "user");

            return message;
        }
    });
}

public static void send(String queueName, String text, int delayMillis) {
    EXECUTOR.submit(() -> {
        try {
            logger.info("*** artificial delay {}: {}", queueName, delayMillis);
            Thread.sleep(delayMillis);
            Connection connection = getConnection();
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            Destination destination = session.createQueue(queueName);
            MessageProducer producer = session.createProducer(destination);
            producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
            TextMessage message = session.createTextMessage(text);
            producer.send(message);
            logger.info("*** sent message {}: {}", queueName, text);
            session.close();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    });
}

public JmsPoolMessageProducer(JmsPoolSession session, MessageProducer messageProducer, Destination destination, boolean shared) throws JMSException {
    this.session = session;
    this.messageProducer = messageProducer;
    this.destination = destination;
    this.shared = shared;
    this.anonymousProducer = destination == null;

    this.deliveryMode = messageProducer.getDeliveryMode();
    this.disableMessageID = messageProducer.getDisableMessageID();
    this.disableMessageTimestamp = messageProducer.getDisableMessageTimestamp();
    this.priority = messageProducer.getPriority();
    this.timeToLive = messageProducer.getTimeToLive();

    if (session.isJMSVersionSupported(2, 0)) {
        this.deliveryDelay = messageProducer.getDeliveryDelay();
    }
}

private void putQueue(List<String> events) throws Exception {
  ConnectionFactory factory = new ActiveMQConnectionFactory(USERNAME,
      PASSWORD, BROKER_BIND_URL);
  Connection connection = factory.createConnection();
  connection.start();

  Session session = connection.createSession(true,
      Session.AUTO_ACKNOWLEDGE);
  Destination destination = session.createQueue(DESTINATION_NAME);
  MessageProducer producer = session.createProducer(destination);

  for (String event : events) {
    TextMessage message = session.createTextMessage();
    message.setText(event);
    producer.send(message);
  }
  session.commit();
  session.close();
  connection.close();
}

public SimpleFilter(String name, String sorgenteMessaggi, String destinazioneMessaggi,
        String connectionFactory, SimpleMessageFilter mf, int maxDelay) {
    this.name = name;
    this.messageSource = (Destination) JndiUtil.getInstance().jndiLookup(sorgenteMessaggi);
    this.messageDestination = (Destination) JndiUtil.getInstance().jndiLookup(destinazioneMessaggi);
    this.connectionFactory = (ConnectionFactory) JndiUtil.getInstance().jndiLookup(connectionFactory);
    this.messageFilter = mf;

    this.maxDelay = maxDelay;

    /* crea un consumatore su sorgenteMessaggi: 
     * girera' messaggi a questo oggetto (this) */
    this.consumer =
            new SimpleAsynchConsumer("Consumatore messaggi per " + this.name,
                    this.messageSource, this.connectionFactory, this, 10);
       logger.info("Creato consumatore: " + consumer.toString());

       /* crea un produttore su destinazioneMessaggi */
    this.producer = new SimpleProducer("Produttore messaggi per " + this.name,
            this.messageDestination, this.connectionFactory, 10);
       logger.info("Creato produttore: " + producer.toString());

       this.messagesReceived = 0;
       this.cancelled = false;
}

private void doSendBinaryMessage( final Session session, final Destination destination,
                                  final byte[] bytes,
                                  final Map<String, ?> properties ) throws JMSException {

    try {
        BytesMessage message = session.createBytesMessage();
        message.writeBytes(bytes);
        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);
    }
}

private void doSend(Destination destination, Message message) throws JMSException {

        if (message == null) {
            throw new MessageFormatException("Message must not be null");
        }

        for (Map.Entry<String, Object> entry : messageProperties.entrySet()) {
            message.setObjectProperty(entry.getKey(), entry.getValue());
        }

        if (correlationId != null) {
            message.setJMSCorrelationID(correlationId);
        }
        if (correlationIdBytes != null) {
            message.setJMSCorrelationIDAsBytes(correlationIdBytes);
        }
        if (type != null) {
            message.setJMSType(type);
        }
        if (replyTo != null) {
            message.setJMSReplyTo(replyTo);
        }

        session.send(producer, destination, message, deliveryMode, priority, timeToLive, disableMessageId, disableTimestamp, deliveryDelay, completionListener);
    }

private void testReception(
        final TestClass i,
        final Destination destination, final ThrowingConsumer<Destination> destinationCheck
) throws Exception {
    final Connection connection = i.connectionFactory.createConnection();
    assertNotNull(connection);
    final Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    assertNotNull(session);
    final MessageProducer producer = session.createProducer(destination);
    assertNotNull(producer);
    final TextMessage message = session.createTextMessage();
    assertNotNull(message);
    message.setText("I am IronMan");
    producer.send(message);

    final List<ReceivedJmsMessage> messages = i.testQueue.drainReceivedMessages();
    assertEquals(1, messages.size());
    final ReceivedJmsMessage receivedMessage = messages.get(0);
    destinationCheck.accept(receivedMessage.getDestination());
    assertTrue(receivedMessage.getJmsMessage() instanceof TextMessage);
    final TextMessage receivedTextMessage = (TextMessage) receivedMessage.getJmsMessage();
    assertEquals("I am IronMan", receivedTextMessage.getText());
}

private MessageProducer getMessageProducer(Destination destination) throws JMSException {
    MessageProducer result = null;

    if (useAnonymousProducers) {
        result = safeGetSessionHolder().getOrCreateProducer();
    } else {
        result = getInternalSession().createProducer(destination);
    }

    return result;
}

@Override
public JMSConsumer createConsumer(Destination destination) {
    try {
        return startIfNeeded(new JmsPoolJMSConsumer((JmsPoolMessageConsumer) getSession().createConsumer(destination)));
    } catch (JMSException jmse) {
        throw JMSExceptionSupport.createRuntimeException(jmse);
    }
}

@Parameters({ "broker-port"})
@Test
public void testConsumerProducerWithAutoAck(String port) throws Exception {
    String queueName = "testConsumerProducerWithAutoAck";
    InitialContext initialContextForQueue = ClientHelper
            .getInitialContextBuilder("admin", "admin", "localhost", port)
            .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(5000);
        Assert.assertNotNull(message, "Message #" + i + " was not received");
    }

    connection.close();
}

@Before
public void setup() throws Exception {
  beforeSetup();
  connectionFactory = mock(ConnectionFactory.class);
  connection = mock(Connection.class);
  session = mock(Session.class);
  queue = mock(Queue.class);
  topic = mock(Topic.class);
  messageConsumer = mock(MessageConsumer.class);
  message = mock(TextMessage.class);
  when(message.getPropertyNames()).thenReturn(new Enumeration<Object>() {
    @Override
    public boolean hasMoreElements() {
      return false;
    }
    @Override
    public Object nextElement() {
      throw new UnsupportedOperationException();
    }
  });
  when(message.getText()).thenReturn(TEXT);
  when(connectionFactory.createConnection(USERNAME, PASSWORD)).thenReturn(connection);
  when(connection.createSession(true, Session.SESSION_TRANSACTED)).thenReturn(session);
  when(session.createQueue(destinationName)).thenReturn(queue);
  when(session.createConsumer(any(Destination.class), anyString())).thenReturn(messageConsumer);
  when(messageConsumer.receiveNoWait()).thenReturn(message);
  when(messageConsumer.receive(anyLong())).thenReturn(message);
  destinationName = DESTINATION_NAME;
  destinationType = JMSDestinationType.QUEUE;
  destinationLocator = JMSDestinationLocator.CDI;
  messageSelector = SELECTOR;
  batchSize = 10;
  pollTimeout = 500L;
  context = new Context();
  converter = new DefaultJMSMessageConverter.Builder().build(context);
  event = converter.convert(message).iterator().next();
  userName = Optional.of(USERNAME);
  password = Optional.of(PASSWORD);
  afterSetup();
}

@Override
public void send(Destination destination, Message message, CompletionListener completionListener)
    throws JMSException {
  Span span = TracingMessageUtils.buildAndInjectSpan(destination, message, tracer);
  messageProducer
      .send(destination, message, new TracingCompletionListener(span, completionListener));
}

@Override
public void send(Destination destination, Message message, int deliveryMode, int priority,
    long timeToLive) throws JMSException {
  Span span = TracingMessageUtils.buildAndInjectSpan(destination, message, tracer);
  try {
    messageProducer.send(destination, message, deliveryMode, priority, timeToLive);
  } catch (Throwable e) {
    SpanJmsDecorator.onError(e, span);
    throw e;
  } finally {
    span.finish();
  }
}

public Message receiveMessage(long timeout, String selector) throws JMSException {
    Connection connection = null;
    Message result = null;
    try {
        connection = startConnection(); //try to be smarter here and start stable 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);
            }
            MessageConsumer consumer;
            if (selector != null) {
                consumer = session.createConsumer(dest, selector);
            } else {
                consumer = session.createConsumer(dest);
            }
            try {
                result = consumer.receive(timeout);
            } finally {
                if (consumer != null) consumer.close();
            }
        } finally {
            if (session != null) session.close();
        }
    } finally {
        safeCloseConnection(connection);
    }
    return result;
}

private void sendJMSMessage(Cargo cargo, Destination destination){
    init();
    jmsCtx.get().createProducer()
    .setPriority(NORMAL_PRIORITY)
    .setDisableMessageID(true)
    .setDisableMessageTimestamp(true)
    .send(destination,
            cargo.getTrackingId().getIdString());

    if(logger.isLoggable(Level.FINEST))
        logger.log(Level.FINEST,"SENT JMS MESSAGE TO " + destination);
}

@Override
public JMSProducer send(Destination destination, Map<String, Object> body) {
    try {
        MapMessage message = session.createMapMessage();
        for (Map.Entry<String, Object> entry : body.entrySet()) {
            message.setObject(entry.getKey(), entry.getValue());
        }

        doSend(destination, message);
    } catch (JMSException jmse) {
        throw JMSExceptionSupport.createRuntimeException(jmse);
    }

    return this;
}

@Override
protected String formatAddress(Destination destination) {
    if (destination == null) {
        return null;
    }
    if (!(destination instanceof ActiveMQDestination)) {
        throw new InvalidParameterException("Destination must be a Core destination, was " + destination.getClass());
    }

    String address = ((ActiveMQDestination) destination).getName();
    return dropDestinationPrefix(address);
}

private void doSend(Destination destination, Message message) throws JMSException {

        if (message == null) {
            throw new MessageFormatException("Message must not be null");
        }

        for (Map.Entry<String, Object> entry : messageProperties.entrySet()) {
            message.setObjectProperty(entry.getKey(), entry.getValue());
        }

        if (correlationId != null) {
            message.setJMSCorrelationID(correlationId);
        }
        if (correlationIdBytes != null) {
            message.setJMSCorrelationIDAsBytes(correlationIdBytes);
        }
        if (type != null) {
            message.setJMSType(type);
        }
        if (replyTo != null) {
            message.setJMSReplyTo(replyTo);
        }

        if (completionListener != null) {
            producer.send(destination, message, deliveryMode, priority, timeToLive, completionListener);
        } else {
            producer.send(destination, message, deliveryMode, priority, timeToLive);
        }
    }

@Override
public MessageConsumer createConsumer(Destination destination, String messageSelector) throws JMSException {
    if (destination instanceof Topic) {
        throw new IllegalStateException("Operation not supported by a QueueSession");
    }
    return super.createConsumer(destination, messageSelector);
}

private void sendTextMessage( final Connection connection, final Destination destination,
                              final String textMessage, final Map<String, ?> properties ) {

    try {
        final Session session = loadSession(false, Session.AUTO_ACKNOWLEDGE);
        doSendTextMessage(session, destination, textMessage, properties);
    } catch (Exception e) {
        throw new JmsMessageException("Failed to send message", e);
    }
}

@Override
public JMSConsumer createConsumer(Destination destination) {
    try {
        return startIfNeeded(new JMSConsumerImpl(this, getSession().createConsumer(destination)));
    } catch (JMSException jmse) {
        throw Utils.convertToRuntimeException(jmse);
    }
}

@Override
public JMSProducer send(Destination destination, byte[] body) {
    try {
        BytesMessage message = context.createBytesMessage();
        message.writeBytes(body);
        doSend(destination, message);
    } catch (JMSException jmse) {
        throw Utils.convertToRuntimeException(jmse);
    }
    return this;
}

/**
 * @see javax.jms.Session#createConsumer(javax.jms.Destination, java.lang.String)
 */
@Override
public MessageConsumer createConsumer(Destination destination, String messageSelector) throws JMSException {
    if (destination instanceof Queue) {
        throw new IllegalStateException("Operation not supported by a TopicSession");
    }
    return super.createConsumer(destination, messageSelector);
}

@Override
public JMSConsumer createConsumer(Destination destination, String selector, boolean noLocal) {
    try {
        return startIfNeeded(new JMSConsumerImpl(this, getSession().createConsumer(destination, selector, noLocal)));
    } catch (JMSException jmse) {
        throw Utils.convertToRuntimeException(jmse);
    }
}

@Override
public ConnectionConsumer createConnectionConsumer(
                                                    Destination destination,
                                                    String messageSelector,
                                                    ServerSessionPool sessionPool,
                                                    int maxMessages ) throws JMSException {

    return connection.createConnectionConsumer(destination, messageSelector, sessionPool, maxMessages);
}

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

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

    session.send(this, destination, message, deliveryMode, priority, timeToLive, disableMessageId, disableTimestamp, deliveryMode, null);
}

/**
 * Inits the.
 *
 * @throws JMSException the jMS exception
 */
public void init() throws JMSException {

    connection = connectionFactory.createConnection();
    connection.start();

    // Create the session
    session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    Destination destination = session.createQueue(queue);
    // Create the producer.
    producer = session.createProducer(destination);

    if (persistent) {
        producer.setDeliveryMode(DeliveryMode.PERSISTENT);
    } else {
        producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
    }
    if (timeToLive != 0) {
        producer.setTimeToLive(timeToLive);
    }
   }

public void sendMessage(MessageContext messageContext,
        Destination destination, Message message) throws JMSException {
    String destinationName = destination.toString();

    if (destination instanceof Topic) {
        messageHandler.sendMessageToTopic(messageContext, destinationName,
                message);
    } else {
        messageHandler.sendMessageToQueue(messageContext, destinationName,
                message);
    }
}

@Override
public void send(Destination destination, Message message) throws JMSException {
  Span span = TracingMessageUtils.buildAndInjectSpan(destination, message, tracer);
  try {
    messageProducer.send(destination, message);
  } catch (Throwable e) {
    SpanJmsDecorator.onError(e, span);
    throw e;
  } finally {
    span.finish();
  }
}

@Override
public JMSConsumer createConsumer(Destination destination) {
    try {
        return startIfNeeded(new MockJMSConsumer(getSession(), (MockJMSMessageConsumer) getSession().createConsumer(destination)));
    } catch (JMSException jmse) {
        throw JMSExceptionSupport.createRuntimeException(jmse);
    }
}

@Override
public JMSConsumer createConsumer(Destination destination, String selector) {
    try {
        return startIfNeeded(new MockJMSConsumer(getSession(), (MockJMSMessageConsumer) getSession().createConsumer(destination, selector)));
    } catch (JMSException jmse) {
        throw JMSExceptionSupport.createRuntimeException(jmse);
    }
}

@Override
public JMSConsumer createConsumer(Destination destination, String selector, boolean noLocal) {
    try {
        return startIfNeeded(new MockJMSConsumer(getSession(), (MockJMSMessageConsumer) getSession().createConsumer(destination, selector, noLocal)));
    } catch (JMSException jmse) {
        throw JMSExceptionSupport.createRuntimeException(jmse);
    }
}

@Override
public JMSProducer send(Destination destination, Message message) {
    try {
        doSend(destination, message);
    } catch (JMSException jmse) {
        throw JMSExceptionSupport.createRuntimeException(jmse);
    }

    return this;
}

@Override
public JMSProducer send(Destination destination, byte[] body) {
    try {
        BytesMessage message = session.createBytesMessage();
        message.writeBytes(body);
        doSend(destination, message);
    } catch (JMSException jmse) {
        throw JMSExceptionSupport.createRuntimeException(jmse);
    }

    return this;
}

@Override
public JMSProducer send(Destination destination, Map<String, Object> body) {
    try {
        MapMessage message = session.createMapMessage();
        for (Map.Entry<String, Object> entry : body.entrySet()) {
            message.setObject(entry.getKey(), entry.getValue());
        }

        doSend(destination, message);
    } catch (JMSException jmse) {
        throw JMSExceptionSupport.createRuntimeException(jmse);
    }

    return this;
}

@Override
public void send(Destination destination, Message message) throws JMSException {
  Span span = TracingMessageUtils.buildAndInjectSpan(destination, message, tracer);
  try {
    messageProducer.send(destination, message);
  } catch (Throwable e) {
    SpanJmsDecorator.onError(e, span);
    throw e;
  } finally {
    span.finish();
  }
}

@Test(timeout = 60000)
public void testRepeatedCreateSessionProducerResultsInSame() throws Exception {
    JmsPoolConnection connection = (JmsPoolConnection) pooledFactory.createConnection();

    assertTrue(pooledFactory.isUseAnonymousProducers());

    Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    Destination destination = session.createTopic("test-topic");
    JmsPoolMessageProducer producer = (JmsPoolMessageProducer) session.createProducer(destination);
    MessageProducer original = producer.getMessageProducer();
    assertNotNull(original);
    session.close();

    assertEquals(1, brokerService.getAdminView().getDynamicDestinationProducers().length);

    for (int i = 0; i < 20; ++i) {
        session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        producer = (JmsPoolMessageProducer) session.createProducer(destination);
        assertSame(original, producer.getMessageProducer());
        session.close();
    }

    assertEquals(1, brokerService.getAdminView().getDynamicDestinationProducers().length);

    connection.close();
    pooledFactory.clear();
}

@Test
public void buildAndInjectSpan() throws Exception {
  Destination destination = new ActiveMQQueue("queue");

  ActiveMQTextMessage message = new ActiveMQTextMessage();
  MockSpan span = mockTracer.buildSpan("test").start();
  mockTracer.scopeManager().activate(span, true);

  MockSpan injected = (MockSpan) TracingMessageUtils
      .buildAndInjectSpan(destination, message, mockTracer);

  assertFalse(message.getProperties().isEmpty());
  assertEquals(span.context().spanId(), injected.parentId());
}