@Override
public long getLong(String name) throws JMSException {
    Object value = getObject(name);

    if (value instanceof Long) {
        return ((Long) value).longValue();
    } else if (value instanceof Integer) {
        return ((Integer) value).longValue();
    } else if (value instanceof Short) {
        return ((Short) value).longValue();
    } else if (value instanceof Byte) {
        return ((Byte) value).longValue();
    } else if (value instanceof String || value == null) {
        return Long.valueOf((String) value).longValue();
    } else {
        throw new MessageFormatException("Cannot read a long from " + value.getClass().getSimpleName());
    }
}
 

@Test
public void testWithJTA() throws JMSException, XAException, InterruptedException {
    TransactionManager transactionManager = new GeronimoTransactionManager();
    Connection connection = createXAConnection("brokerJTA", transactionManager);
    Queue dest = JMSUtil.createQueue(connection, "test");

    MessageListener listenerHandler = new TestMessageListener();
    ExceptionListener exListener = new TestExceptionListener();
    PollingMessageListenerContainer container = new PollingMessageListenerContainer(connection, dest,
                                                                                    listenerHandler, exListener);
    container.setTransacted(false);
    container.setAcknowledgeMode(Session.SESSION_TRANSACTED);
    container.setTransactionManager(transactionManager);
    container.start();

    testTransactionalBehaviour(connection, dest);

    container.stop();
    connection.close();
}
 

@Test
public void testWithQueueConnectionFactoryAndJms11Usage() throws JMSException {
	QueueConnectionFactory cf = mock(QueueConnectionFactory.class);
	QueueConnection con = mock(QueueConnection.class);

	given(cf.createConnection()).willReturn(con);

	SingleConnectionFactory scf = new SingleConnectionFactory(cf);
	Connection con1 = scf.createConnection();
	Connection con2 = scf.createConnection();
	con1.start();
	con2.start();
	con1.close();
	con2.close();
	scf.destroy();  // should trigger actual close

	verify(con).start();
	verify(con).stop();
	verify(con).close();
	verifyNoMoreInteractions(con);
}
 

@Test
public void testConnectionProblemXA() throws JMSException, XAException, InterruptedException {
    TransactionManager transactionManager = new GeronimoTransactionManager();
    Connection connection = createXAConnection("brokerJTA", transactionManager);
    Queue dest = JMSUtil.createQueue(connection, "test");

    MessageListener listenerHandler = new TestMessageListener();
    TestExceptionListener exListener = new TestExceptionListener();

    PollingMessageListenerContainer container = //
        new PollingMessageListenerContainer(connection, dest, listenerHandler, exListener);
    container.setTransacted(false);
    container.setAcknowledgeMode(Session.SESSION_TRANSACTED);
    container.setTransactionManager(transactionManager);

    connection.close(); // Simulate connection problem
    container.start();
    Awaitility.await().until(() -> exListener.exception != null);
    JMSException ex = exListener.exception;
    assertNotNull(ex);
    // Closing the pooled connection will result in a NPE when using it
    assertEquals("Wrapped exception. null", ex.getMessage());
}
 

@Override
public void deleteMessage(String messageId) throws ListenerException {
	Session session=null;
	MessageConsumer mc = null;
	try {
		session = createSession();
		log.debug("retrieving message ["+messageId+"] in order to delete it");
		mc = getMessageConsumer(session, getDestination(), getCombinedSelector(messageId));
		mc.receive(getTimeOut());
	} catch (Exception e) {
		throw new ListenerException(e);
	} finally {
		try {
			if (mc != null) {
				mc.close();
			}
		} catch (JMSException e1) {
			throw new ListenerException("exception closing message consumer",e1);
		}
		closeSession(session);
	}
}
 

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

@Test
public void testRuntimeExceptionOnCreateMapMessage() throws JMSException {
    JmsConnection connection = Mockito.mock(JmsConnection.class);
    JmsSession session = Mockito.mock(JmsSession.class);
    Mockito.when(connection.createSession(Mockito.anyInt())).thenReturn(session);
    JmsContext context = new JmsContext(connection, JMSContext.CLIENT_ACKNOWLEDGE);

    Mockito.doThrow(IllegalStateException.class).when(session).createMapMessage();

    try {
        context.createMapMessage();
        fail("Should throw ISRE");
    } catch (IllegalStateRuntimeException isre) {
    } finally {
        context.close();
    }
}
 

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

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

private MapMessage createLargeMessage() throws JMSException {
   MapMessage message = session.createMapMessage();

   for (int i = 0; i < 10; i++) {
      message.setBytes("test" + i, new byte[1024 * 1024]);
   }
   return message;
}
 

/**
 * Test send when destination already specified
 */
@Test
public void testSendDestinationAlreadySpecified() throws JMSException {

    SQSTextMessage msg = spy(new SQSTextMessage("MyText"));

    try {
        producer.send(destination, msg);
        fail();
    } catch (UnsupportedOperationException ide) {
        // expected
    }

    verify(producer).checkIfDestinationAlreadySet();
}
 

public void createTopic(String topicName) throws JMSException {
    if (StringUtils.isBlank(topicName)) {
        throw WeEventConnectionFactory.error2JMSException(ErrorCode.TOPIC_IS_BLANK);
    }
    try {
        this.client.open(topicName);
    } catch (BrokerException e) {
        log.info("create topic error.", e);
        throw WeEventConnectionFactory.exp2JMSException(e);
    }
}
 

/**
 * Determine whether there are currently thread-bound credentials,
 * using them if available, falling back to the statically specified
 * username and password (i.e. values of the bean properties) else.
 * @see #doCreateConnection
 */
@Override
public final Connection createConnection() throws JMSException {
	JmsUserCredentials threadCredentials = this.threadBoundCredentials.get();
	if (threadCredentials != null) {
		return doCreateConnection(threadCredentials.username, threadCredentials.password);
	}
	else {
		return doCreateConnection(this.username, this.password);
	}
}
 

@Test(timeout = 20000)
public void testOnExceptionFiredOnSessionPoolFailure() throws Exception {
    final CountDownLatch exceptionFired = new CountDownLatch(1);

    try (TestAmqpPeer testPeer = new TestAmqpPeer();) {
        Connection connection = testFixture.establishConnecton(testPeer);
        connection.setExceptionListener(new ExceptionListener() {

            @Override
            public void onException(JMSException exception) {
                exceptionFired.countDown();
            }
        });

        connection.start();

        JmsFailingServerSessionPool sessionPool = new JmsFailingServerSessionPool();

        // Now the Connection consumer arrives and we give it a message
        // to be dispatched to the server session.
        DescribedType amqpValueNullContent = new AmqpValueDescribedType(null);

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

        Queue queue = new JmsQueue("myQueue");
        ConnectionConsumer consumer = connection.createConnectionConsumer(queue, null, sessionPool, 100);

        assertTrue("Exception should have been fired", exceptionFired.await(5, TimeUnit.SECONDS));

        testPeer.expectDetach(true, true, true);
        testPeer.expectDispositionThatIsReleasedAndSettled();
        consumer.close();

        testPeer.expectClose();
        connection.close();

        testPeer.waitForAllHandlersToComplete(1000);
    }
}
 

@Override
public void publish(final Message message,
                    final int deliveryMode,
                    final int priority,
                    final long timeToLive) throws JMSException {
   send(message, deliveryMode, priority, timeToLive);
}
 

@Override
public TopicSession createTopicSession(boolean transacted, int acknowledgeMode) throws JMSException {
    checkClosedOrFailed();
    ensureConnected();
    int ackMode = getSessionAcknowledgeMode(transacted, acknowledgeMode);
    MockJMSTopicSession result = new MockJMSTopicSession(getNextSessionId(), ackMode, this);
    addSession(result);
    if (started.get()) {
        result.start();
    }
    return result;
}
 

/**
 * if a property is set as a <code>byte</code>,
 * it can also be read as a <code>long</code>.
 */
@Test
public void testByte2Long() {
   try {
      Message message = senderSession.createMessage();
      message.setByteProperty("prop", (byte) 127);
      Assert.assertEquals(127L, message.getLongProperty("prop"));
   } catch (JMSException e) {
      fail(e);
   }
}
 

/**
 * Publish messages to a topic in a node and receive from the same node at a slow rate.
 *
 * @throws AndesEventAdminServiceEventAdminException
 * @throws org.wso2.mb.integration.common.clients.exceptions.AndesClientConfigurationException
 * @throws XPathExpressionException
 * @throws NamingException
 * @throws JMSException
 * @throws IOException
 */
@Test(groups = "wso2.mb", description = "Same node publisher, slow subscriber test case",
        enabled = true)
@Parameters({"messageCount"})
public void testSameNodeSlowSubscriber(long messageCount)
        throws AndesEventAdminServiceEventAdminException, AndesClientConfigurationException,
               XPathExpressionException, NamingException, JMSException, IOException, AndesClientException,
               DataAccessUtilException {
    this.runSingleSubscriberSinglePublisherTopicTestCase(
                    automationContextForMB2, automationContextForMB2, 10L, 0L, "singleTopic2", messageCount, true);
}
 

@Override
public Object getProperty(AmqpJmsMessageFacade message) throws JMSException {
    if (message instanceof AmqpJmsObjectMessageFacade) {
        return ((AmqpJmsObjectMessageFacade) message).isAmqpTypedEncoding();
    }

    return null;
}
 

/**
 * Gets a String valued property from a JMS message.
 *
 * @param message The message.
 * @param name The property name.
 * @return The property value or {@code null} if the message does not contain the corresponding property.
 */
public static String getStringProperty(final Message message, final String name)  {
    try {
        return message.getStringProperty(name);
    } catch (final JMSException e) {
        return null;
    }
}
 

public static void close(final Connection closeable) {
    if (closeable != null) {
        try {
            closeable.close();
        } catch (final JMSException e) {
        }
    }
}
 

/**
 * Apply vendor-specific settings for building up a connection factory to
 * WMQ.
 * 
 * @param cf
 * @throws JMSException
 */
protected void configureMQConnectionFactory(MQConnectionFactory cf)
		throws JMSException {

	// always client bindings
	cf.setTransportType(JMSC.MQJMS_TP_CLIENT_MQ_TCPIP);
	cf.setHostName(Config.parms.getString("cmd_jh"));
	cf.setPort(Config.parms.getInt("cmd_p"));
	cf.setChannel(Config.parms.getString("cmd_jc"));

	cf.setQueueManager(Config.parms.getString("cmd_jb"));

}
 

@Override
public synchronized void stop() throws JMSException
{
    int count = startCount.decrementAndGet();
    if (count == 0)
    {
        underlying.stop();
    }
}
 

@Override
public byte[] getBytes(final String name) throws JMSException {
   try {
      return map.getBytesProperty(new SimpleString(name));
   } catch (ActiveMQPropertyConversionException e) {
      throw new MessageFormatException(e.getMessage());
   }
}
 

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

@Override
public TopicSession createTopicSession(
                                        boolean transacted,
                                        int acknowledgeMode ) throws JMSException {

    return addSession( ((TopicConnection) connection).createTopicSession(transacted, acknowledgeMode));
}
 

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

public void testSetReadOnly() {
   ActiveMQMessage msg = new ActiveMQMessage();
   msg.setReadOnlyProperties(true);
   boolean test = false;
   try {
      msg.setIntProperty("test", 1);
   } catch (MessageNotWriteableException me) {
      test = true;
   } catch (JMSException e) {
      e.printStackTrace(System.err);
      test = false;
   }
   assertTrue(test);
}
 

protected MessageConsumer createConsumer() throws JMSException {
   if (durable) {
      LOG.info("Creating durable consumer");
      return session.createDurableSubscriber((Topic) consumerDestination, getName());
   }
   return session.createConsumer(consumerDestination);
}
 

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

    jmsTemplate.send(destinationName, new MessageCreator() {
        @Override
        public Message createMessage(Session session) throws JMSException {
            TextMessage message = session.createTextMessage("hello from the other side");
            message.setStringProperty("foo", "foo");
            message.setBooleanProperty("bar", false);
            message.setJMSReplyTo(session.createQueue("fooQueue"));
            return message;
        }
    });

    JMSConsumer consumer = new JMSConsumer(jmsTemplate, mock(ComponentLog.class));
    final AtomicBoolean callbackInvoked = new AtomicBoolean();
    consumer.consume(destinationName, new ConsumerCallback() {
        @Override
        public void accept(JMSResponse response) {
            callbackInvoked.set(true);
            assertEquals("hello from the other side", new String(response.getMessageBody()));
            assertEquals("fooQueue", response.getMessageHeaders().get(JmsHeaders.REPLY_TO));
            assertEquals("foo", response.getMessageProperties().get("foo"));
            assertEquals("false", response.getMessageProperties().get("bar"));
        }
    });
    assertTrue(callbackInvoked.get());

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

public void testSetJMSDeliveryModeProperty() throws JMSException {
   ActiveMQMessage message = new ActiveMQMessage();
   String propertyName = "JMSDeliveryMode";

   // Set as Boolean
   message.setObjectProperty(propertyName, Boolean.TRUE);
   assertTrue(message.isPersistent());
   message.setObjectProperty(propertyName, Boolean.FALSE);
   assertFalse(message.isPersistent());
   message.setBooleanProperty(propertyName, true);
   assertTrue(message.isPersistent());
   message.setBooleanProperty(propertyName, false);
   assertFalse(message.isPersistent());

   // Set as Integer
   message.setObjectProperty(propertyName, DeliveryMode.PERSISTENT);
   assertTrue(message.isPersistent());
   message.setObjectProperty(propertyName, DeliveryMode.NON_PERSISTENT);
   assertFalse(message.isPersistent());
   message.setIntProperty(propertyName, DeliveryMode.PERSISTENT);
   assertTrue(message.isPersistent());
   message.setIntProperty(propertyName, DeliveryMode.NON_PERSISTENT);
   assertFalse(message.isPersistent());

   // Set as String
   message.setObjectProperty(propertyName, "PERSISTENT");
   assertTrue(message.isPersistent());
   message.setObjectProperty(propertyName, "NON_PERSISTENT");
   assertFalse(message.isPersistent());
   message.setStringProperty(propertyName, "PERSISTENT");
   assertTrue(message.isPersistent());
   message.setStringProperty(propertyName, "NON_PERSISTENT");
   assertFalse(message.isPersistent());
}