Java 类javax.jms.Queue 实例源码

@BeforeEach
public void setUp() throws JMSException {
    initMocks(this);
    when(mockEvent.getTarget()).thenReturn(of(rdf.createIRI("trellis:repository/resource")));
    when(mockEvent.getAgents()).thenReturn(singleton(Trellis.AdministratorAgent));
    when(mockEvent.getCreated()).thenReturn(time);
    when(mockEvent.getIdentifier()).thenReturn(rdf.createIRI("urn:test"));
    when(mockEvent.getTypes()).thenReturn(singleton(AS.Update));
    when(mockEvent.getTargetTypes()).thenReturn(singleton(LDP.RDFSource));
    when(mockEvent.getInbox()).thenReturn(empty());

    when(mockConnection.createSession(anyBoolean(), eq(AUTO_ACKNOWLEDGE))).thenReturn(mockSession);
    when(mockSession.createQueue(eq(queueName))).thenReturn(mockQueue);
    when(mockSession.createTextMessage(anyString())).thenReturn(mockMessage);
    when(mockSession.createProducer(any(Queue.class))).thenReturn(mockProducer);

    doNothing().when(mockProducer).send(any(TextMessage.class));
}

/**
 * Deals with reconnecting or if broker gone down, fails
 *
 * @param queueName
 * @return
 * @throws JMSException
 */
protected Queue createQueue(String queueName) throws JMSException {

    // Deals with reconnecting or if broker gone down, fails
    try {
        if (connection==null) createConnection();
        if (qSession == null) createQSession();

        return qSession.createQueue(queueName);

    } catch (Exception ne) {
        createConnection();
        createQSession();

        return qSession.createQueue(queueName);
    }
}

public String sendRequest(Optional<String> routeId) {
    CompactRequest req = new CompactRequest(routeId.orElse("asdf"));
    try {
        TextMessage msg = context.createTextMessage(JsonMapper.serializeOrThrow(req));
        msg.setJMSDeliveryMode(DeliveryMode.NON_PERSISTENT);

        Queue answerQ = context.createTemporaryQueue();
        msg.setJMSReplyTo(answerQ);

        context.createProducer().send(minQ, msg);

        Message response = context.createConsumer(answerQ).receive();
        if (response instanceof TextMessage) {
            return ((TextMessage) response).getText();
        }

        return "";
    } catch (JMSException e) {
        return e.getMessage();
    }
}

@Test
public void testReceiveNoWait() throws JMSException {
    JmsPoolConnection connection = (JmsPoolConnection) cf.createQueueConnection();
    Session session = connection.createSession();
    Queue queue = session.createTemporaryQueue();
    MessageConsumer consumer = session.createConsumer(queue, "Color = Red");

    assertNull(consumer.receiveNoWait());

    consumer.close();

    try {
        consumer.receiveNoWait();
        fail("Should not be able to interact with closed consumer");
    } catch (IllegalStateException ise) {}
}

private void doTestCreateQueueSender(boolean useAnonymousProducers) throws JMSException {
    cf.setUseAnonymousProducers(useAnonymousProducers);

    JmsPoolConnection connection = (JmsPoolConnection) cf.createConnection();
    QueueSession session = connection.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);

    Queue queue1 = session.createTemporaryQueue();
    Queue queue2 = session.createTemporaryQueue();

    JmsPoolQueueSender sender1 = (JmsPoolQueueSender) session.createSender(queue1);
    JmsPoolQueueSender sender2 = (JmsPoolQueueSender) session.createSender(queue2);

    if (useAnonymousProducers) {
        assertSame(sender1.getMessageProducer(), sender2.getMessageProducer());
    } else {
        assertNotSame(sender1.getMessageProducer(), sender2.getMessageProducer());
    }

    connection.close();
}

@Test
public void testGetTopicSubscriber() throws JMSException {
    JmsPoolConnection connection = (JmsPoolConnection) cf.createQueueConnection();
    QueueSession session = connection.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
    Queue queue = session.createTemporaryQueue();
    JmsPoolQueueSender sender = (JmsPoolQueueSender) session.createSender(queue);

    assertNotNull(sender.getQueueSender());
    assertTrue(sender.getQueueSender() instanceof MockJMSQueueSender);

    sender.close();

    try {
        sender.getQueueSender();
        fail("Cannot read state on closed sender");
    } catch (IllegalStateException ise) {}
}

@Test
public void testGetMessageSelector() throws JMSException {
    JmsPoolConnection connection = (JmsPoolConnection) cf.createQueueConnection();
    QueueSession session = connection.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
    Queue queue = session.createTemporaryQueue();
    QueueBrowser browser = session.createBrowser(queue, "color = red");

    assertNotNull(browser.getMessageSelector());
    assertEquals("color = red", browser.getMessageSelector());

    browser.close();

    try {
        browser.getMessageSelector();
        fail("Should not be able to use a closed browser");
    } catch (IllegalStateException ise) {
    }
}

@Test
public void testGetEnumeration() throws JMSException {
    JmsPoolConnection connection = (JmsPoolConnection) cf.createQueueConnection();
    QueueSession session = connection.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
    Queue queue = session.createTemporaryQueue();
    QueueBrowser browser = session.createBrowser(queue);

    assertNotNull(browser.getEnumeration());

    browser.close();

    try {
        browser.getEnumeration();
        fail("Should not be able to use a closed browser");
    } catch (IllegalStateException ise) {
    }
}

public String sendRequest(Optional<String> routeId) {
    DetailsRequest req = new DetailsRequest(routeId.orElse("asdf"));
    try {
        TextMessage msg = context.createTextMessage(JsonMapper.serializeOrThrow(req));
        msg.setJMSDeliveryMode(DeliveryMode.NON_PERSISTENT);

        Queue answerQ = context.createTemporaryQueue();
        msg.setJMSReplyTo(answerQ);

        context.createProducer().send(minQ, msg);

        Message response = context.createConsumer(answerQ).receive();
        if (response instanceof TextMessage) {
            return ((TextMessage) response).getText();
        }

        return "";
    } catch (JMSException e) {
        return e.getMessage();
    }
}

@Override
public void run(String... strings) throws Exception {
    final String messageText = "Hello World";
    LOG.info("============= Sending " + messageText);
    this.jmsTemplate.send("example", new MessageCreator() {

        @Override
        public Message createMessage(Session session) throws JMSException {
            Queue replyQueue = session.createQueue("reply-queue");

            TextMessage message = session.createTextMessage(messageText);

            message.setJMSCorrelationID(correlationID.toString());
            message.setJMSReplyTo(replyQueue);

            return message;
        }
    });
}

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

  connection = connFactory.createConnection();
  // lets make it transactional
  session = connection.createSession(true, Session.SESSION_TRANSACTED);
  Queue controllerQueue = session.createQueue(ctrlrQueueName);

  MessageConsumer consumer = session.createConsumer(controllerQueue);

  consumer.setMessageListener(this);
  connection.start();

  logger.info(">>>>>>>>>>>>>>>>Inductor Listener Waiting for messages...");
}

private QueueSender getQueueSender(Queue destination) throws JMSException {
    QueueSender result = null;

    if (useAnonymousProducers) {
        result = safeGetSessionHolder().getOrCreateSender();
    } else {
        result = ((QueueSession) getInternalSession()).createSender(destination);
    }

    return result;
}

@Override
public Queue createQueue(String queueName) {
    try {
        return getSession().createQueue(queueName);
    } catch (JMSException jmse) {
        throw JMSExceptionSupport.createRuntimeException(jmse);
    }
}

@Override
public QueueBrowser createBrowser(Queue queue) {
    try {
        return startIfNeeded(getSession().createBrowser(queue));
    } catch (JMSException jmse) {
        throw JMSExceptionSupport.createRuntimeException(jmse);
    }
}

@Override
public QueueBrowser createBrowser(Queue queue, String selector) {
    try {
        return startIfNeeded(getSession().createBrowser(queue, selector));
    } catch (JMSException jmse) {
        throw JMSExceptionSupport.createRuntimeException(jmse);
    }
}

/**
 * Get the OSCM task queue
 * 
 * @param context
 *            a JNDI context
 * @return the task queue
 */
private Queue getJmsQueue(InitialContext context) {
    try {
        Object lookup = context.lookup(JMS_QUEUE_JNDI_NAME);
        return Queue.class.cast(lookup);
    } catch (Exception e) {
        e.printStackTrace();
        return null;
    }
}

@Test
public void testToString() throws JMSException {
    JmsPoolConnection connection = (JmsPoolConnection) cf.createQueueConnection();
    Session session = connection.createSession();
    Queue queue = session.createTemporaryQueue();
    MessageProducer producer = session.createProducer(queue);

    assertNotNull(producer.toString());
}

@Test
public void testCloseMoreThanOnce() throws JMSException {
    JmsPoolConnection connection = (JmsPoolConnection) cf.createQueueConnection();
    Session session = connection.createSession();
    Queue queue = session.createTemporaryQueue();
    MessageProducer producer = session.createProducer(queue);

    producer.close();
    producer.close();
}

@Override
public QueueReceiver createReceiver(
                                     Queue queue,
                                     String messageSelector ) throws JMSException {

    return addConsumer( ((QueueSession) session).createReceiver(queue, messageSelector));
}

static void queueMessage(Session session, String text, String queueName) {
    try {
        Queue queue = session.createQueue(queueName);
        MessageProducer producer = session.createProducer(queue);

        TextMessage message = session.createTextMessage(text);
        message.setJMSCorrelationID(MDC.get(X_OBOS_REQUEST_ID));

        producer.send(message);
        session.commit();
    } catch (JMSException ex) {
        throw new MessageQueueException("Could not queue message '" + text + "'", ex);
    }
}

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

/**
 * @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 void requeueFailedMessages() {
    try {
        ActiveMQConnection connection = ActiveMqUtils.openConnection(user, password, url);
        Session session = ActiveMqUtils.startSession(connection);

        int count = getQueueSize(session, queueError);

        if (count < 1) {
            return;
        }

        log.info("Requeuing {} failed messages...", count);

        Queue queueErr = session.createQueue(queueError);
        MessageConsumer consumer = session.createConsumer(queueErr);

        Queue queueRetry = session.createQueue(queueInput);
        MessageProducer producer = session.createProducer(queueRetry);

        for (int consumed = 0; consumed < count; consumed++) {
            TextMessage message = (TextMessage) consumer.receive(REQUEUE_TIMEOUT);

            if (message == null) {
                continue;
            }

            String text = message.getText();
            String requestId = message.getJMSCorrelationID();

            log.info("Requeuing message '{}'", text);

            try {
                TextMessage newMessage = session.createTextMessage(text);
                newMessage.setJMSCorrelationID(requestId);

                producer.send(newMessage);
            } catch (Exception e) {
                log.error("Failed to requeue message", e);
            }

            message.acknowledge();
            session.commit();
        }

        producer.close();
        consumer.close();
    } catch (JMSException ex) {
        throw new MessageQueueException("Failed to requeue failed messages", ex);
    }
}

private List<String> consumeMessages(JMSContext context, Queue queue) {
    List<String> messages = new ArrayList<>();
    try (JMSConsumer consumer = context.createConsumer(queue)) {
        while (true) {
            String msg = consumer.receiveBody(String.class, 100);
            if (msg != null) {
                messages.add(msg);
            } else {
                return messages;
            }
        }
    }
}

@Test
public void validateSuccessfulPublishAndTransferToSuccess() throws Exception {
    ActiveMQConnectionFactory cf = new ActiveMQConnectionFactory("vm://localhost?broker.persistent=false");

    final String destinationName = "fooQueue";
    PublishJMS pubProc = new PublishJMS();
    TestRunner runner = TestRunners.newTestRunner(pubProc);
    JMSConnectionFactoryProviderDefinition cs = mock(JMSConnectionFactoryProviderDefinition.class);
    when(cs.getIdentifier()).thenReturn("cfProvider");
    when(cs.getConnectionFactory()).thenReturn(cf);

    runner.addControllerService("cfProvider", cs);
    runner.enableControllerService(cs);

    runner.setProperty(PublishJMS.CF_SERVICE, "cfProvider");
    runner.setProperty(PublishJMS.DESTINATION, destinationName);

    Map<String, String> attributes = new HashMap<>();
    attributes.put("foo", "foo");
    attributes.put(JmsHeaders.REPLY_TO, "cooQueue");
    runner.enqueue("Hey dude!".getBytes(), attributes);
    runner.run(1, false);

    final MockFlowFile successFF = runner.getFlowFilesForRelationship(PublishJMS.REL_SUCCESS).get(0);
    assertNotNull(successFF);

    JmsTemplate jmst = new JmsTemplate(cf);
    BytesMessage message = (BytesMessage) jmst.receive(destinationName);

    byte[] messageBytes = MessageBodyToBytesConverter.toBytes(message);
    assertEquals("Hey dude!", new String(messageBytes));
    assertEquals("cooQueue", ((Queue) message.getJMSReplyTo()).getQueueName());
    assertEquals("foo", message.getStringProperty("foo"));
}

@Parameters({"admin-username", "admin-password", "broker-hostname", "broker-port"})
@Test
public void testRetrieveConsumerList(String username, String password,
                                     String hostname, String port) throws Exception {
    String queueName = "testSpecificQueueRetrieval";


    // Create a durable queue using a JMS client
    InitialContext initialContextForQueue = ClientHelper
            .getInitialContextBuilder(username, password, hostname, port)
            .withQueue(queueName)
            .build();

    QueueConnectionFactory connectionFactory
            = (QueueConnectionFactory) initialContextForQueue.lookup(ClientHelper.CONNECTION_FACTORY);
    QueueConnection connection = connectionFactory.createQueueConnection();
    connection.start();

    QueueSession queueSession = connection.createQueueSession(false, QueueSession.AUTO_ACKNOWLEDGE);
    Queue queue = queueSession.createQueue(queueName);
    QueueReceiver receiver1 = queueSession.createReceiver(queue);
    QueueReceiver receiver2 = queueSession.createReceiver(queue);

    HttpGet httpGet = new HttpGet(apiBasePath + QueuesApiDelegate.QUEUES_API_PATH
                                          + "/" + queueName + "/consumers");

    CloseableHttpResponse response = client.execute(httpGet);
    String body = EntityUtils.toString(response.getEntity());

    ConsumerMetadata[] consumers = objectMapper.readValue(body, ConsumerMetadata[].class);

    Assert.assertEquals(consumers.length, 2, "Number of consumers returned is incorrect.");

    receiver1.close();
    receiver2.close();
    queueSession.close();
    connection.close();
}

@Test
public void testToString() throws JMSException {
    JmsPoolConnection connection = (JmsPoolConnection) cf.createQueueConnection();
    Session session = connection.createSession();
    Queue queue = session.createTemporaryQueue();
    MessageConsumer consumer = session.createConsumer(queue);

    assertNotNull(consumer.toString());
}

@Test
public void testCloseMoreThanOnce() throws JMSException {
    JmsPoolConnection connection = (JmsPoolConnection) cf.createQueueConnection();
    Session session = connection.createSession();
    Queue queue = session.createTemporaryQueue();
    MessageConsumer consumer = session.createConsumer(queue);

    consumer.close();
    consumer.close();
}

/** Connect to the server, and lookup the managed resources. 
 * @throws JMSException */
public void initialize() throws NamingException {
    System.out.println("Getting the InitialContext");
    InitialContext context = new InitialContext();

    //lookup our JMS objects
    System.out.println("Looking up our JMS resources");
    queueCF = (QueueConnectionFactory) context.lookup(NOTIFICATION_QCF);
    queue = (Queue) context.lookup(NOTIFICATION_Q);

    initialized = true;
    System.out.println("Initialization completed successfully!");
}

@Test
public void testSimpleSubmission() throws Exception {

    StatusBean bean = doSubmit();

    // Manually take the submission from the list not using event service for isolated test
    ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(submitter.getUri());
    Connection connection = connectionFactory.createConnection();

    try {
        Session   session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        Queue queue = session.createQueue(IEventService.SUBMISSION_QUEUE);

        final MessageConsumer consumer = session.createConsumer(queue);
        connection.start();

        TextMessage msg = (TextMessage)consumer.receive(1000);

        IMarshallerService marshaller = new MarshallerService(new PointsModelMarshaller());
        StatusBean fromQ = marshaller.unmarshal(msg.getText(), StatusBean.class);

    if (!fromQ.equals(bean)) throw new Exception("The bean from the queue was not the same as that submitted! q="+fromQ+" submit="+bean);

    } finally {
        consumer.disconnect();
        connection.close();
    }
}

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

    return addConnectionConsumer(queueConnection.createConnectionConsumer(queue,
                                                                          messageSelector,
                                                                          sessionPool,
                                                                          maxMessages));
}

@Test(timeout = 60000)
public void testCreateReceiverWithSelector() throws Exception {
    JmsPoolConnection connection = (JmsPoolConnection) cf.createConnection();
    JmsPoolSession session = (JmsPoolSession) connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

    Queue queue = session.createQueue(getTestName());
    assertNotNull(session.createReceiver(queue, "color = red"));

    session.close();
    try {
        session.createReceiver(queue, "color = greean");
        fail("Should not be able to createReceiver when closed");
    } catch (JMSException ex) {}
}

@Test
public void testToString() throws JMSException {
    JmsPoolConnection connection = (JmsPoolConnection) cf.createQueueConnection();
    QueueSession session = connection.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
    Queue queue = session.createTemporaryQueue();
    QueueSender sender = session.createSender(queue);

    assertNotNull(sender.toString());
}

@Override
public QueueReceiver createSubscriber() throws JMSException
{
    QueueReceiver recv = ((QueueSession) session).createReceiver((Queue) topic, messageSelector);
    log.debug("Created non-durable subscriber");
    return recv;
}

/**
 *
 */
private String retrieveDestinationName(Destination destination, String headerName) {
    String destinationName = null;
    if (destination != null) {
        try {
            destinationName = (destination instanceof Queue) ? ((Queue) destination).getQueueName()
                    : ((Topic) destination).getTopicName();
        } catch (JMSException e) {
            this.processLog.warn("Failed to retrieve Destination name for '" + headerName + "' header", e);
        }
    }
    return destinationName;
}

@Test(timeout = 30000)
public void testCreateConsumerWithSelector() {
    JMSContext context = cf.createContext();
    Queue queue = context.createQueue(getTestName());
    assertNotNull(context.createConsumer(queue, "color = red"));

    context.close();
    try {
        context.createConsumer(queue, "color = blue");
        fail("Should not be able to create resource when context is closed");
    } catch (IllegalStateRuntimeException isre) {}
}

@Test(timeout = 60000)
public void testTemporaryQueueWithMultipleConnectionUsers() throws Exception {
    JmsPoolConnection connection1 = null;
    JmsPoolConnection connection2 = null;

    MockJMSConnection pooledConnection = null;

    Session session1 = null;
    Session session2 = null;
    Queue tempQueue = null;
    Queue normalQueue = null;

    connection1 = (JmsPoolConnection) cf.createConnection();
    pooledConnection = (MockJMSConnection) connection1.getConnection();
    session1 = connection1.createSession(false, Session.AUTO_ACKNOWLEDGE);
    tempQueue = session1.createTemporaryQueue();
    LOG.info("Created temporary queue named: " + tempQueue.getQueueName());

    assertEquals(1, pooledConnection.getConnectionStats().getActiveTemporaryQueueCount());

    connection2 = (JmsPoolConnection) cf.createConnection();
    assertSame(connection1.getConnection(), connection2.getConnection());
    session2 = connection2.createSession(false, Session.AUTO_ACKNOWLEDGE);
    normalQueue = session2.createQueue("queue:FOO.TEST");
    LOG.info("Created queue named: " + normalQueue.getQueueName());

    // didn't create a temp queue on pooledConnection2 so we should still have a temp queue
    connection2.close();
    assertEquals(1, pooledConnection.getConnectionStats().getActiveTemporaryQueueCount());

    // after closing pooledConnection, where we created the temp queue, there should
    // be no temp queues left
    connection1.close();
    assertEquals(0, pooledConnection.getConnectionStats().getActiveTemporaryQueueCount());
}

@Test
public void testToString() throws JMSException {
    JmsPoolConnection connection = (JmsPoolConnection) cf.createQueueConnection();
    QueueSession session = connection.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
    Queue queue = session.createTemporaryQueue();
    QueueBrowser browser = session.createBrowser(queue);

    assertNotNull(browser.toString());
}

@Test
public void validateSuccessfulPublishAndTransferToSuccessWithELOverJNDI() throws Exception {
    ActiveMQConnectionFactory cf = (ActiveMQConnectionFactory) CommonTest.buildJmsJndiConnectionFactory();

    final String destinationNameExpression = "${foo}Queue";
    final String destinationName = "fooQueue";
    PublishJMS pubProc = new PublishJMS();
    TestRunner runner = TestRunners.newTestRunner(pubProc);
    JMSConnectionFactoryProviderDefinition cs = mock(JMSConnectionFactoryProviderDefinition.class);
    when(cs.getIdentifier()).thenReturn("cfProvider");
    when(cs.getConnectionFactory()).thenReturn(cf);

    runner.addControllerService("cfProvider", cs);
    runner.enableControllerService(cs);

    runner.setProperty(PublishJMS.CF_SERVICE, "cfProvider");
    runner.setProperty(PublishJMS.DESTINATION, destinationNameExpression);

    Map<String, String> attributes = new HashMap<>();
    attributes.put("foo", "foo");
    attributes.put(JmsHeaders.REPLY_TO, "cooQueue");
    runner.enqueue("Hey dude!".getBytes(), attributes);
    runner.run(1, false);

    final MockFlowFile successFF = runner.getFlowFilesForRelationship(PublishJMS.REL_SUCCESS).get(0);
    assertNotNull(successFF);

    JmsTemplate jmst = new JmsTemplate(cf);
    BytesMessage message = (BytesMessage) jmst.receive(destinationName);

    byte[] messageBytes = MessageBodyToBytesConverter.toBytes(message);
    assertEquals("Hey dude!", new String(messageBytes));
    assertEquals("cooQueue", ((Queue) message.getJMSReplyTo()).getQueueName());
    assertEquals("foo", message.getStringProperty("foo"));
}

@Test
public void testFailedCreateConsumerConnectionStillWorks() throws JMSException {
    // User can write but not read
    user.setCanConsumeAll(false);

    Connection connection = null;

    try {
        connection = cf.createConnection("admin", "admin");
    } catch (JMSSecurityException jmsse) {
        fail("Should not be able to create connection using bad credentials");
    }

    connection.start();

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

    try {
        session.createConsumer(queue);
        fail("Should fail to create consumer");
    } catch (JMSSecurityException ex) {
        LOG.debug("Caught expected security error");
    }

    MessageProducer producer = session.createProducer(queue);
    producer.close();

    connection.close();
}