@Override
public JMSContext createContext(String username, String password, int sessionMode) {
    if (stopped.get()) {
        LOG.debug("JmsPoolConnectionFactory is stopped, skip create new connection.");
        return null;
    }

    if (!jmsContextSupported) {
        throw new JMSRuntimeException("Configured ConnectionFactory is not JMS 2+ capable");
    }

    if (isUseProviderJMSContext()) {
        return createProviderContext(username, password, sessionMode);
    } else {
        try {
            return newPooledConnectionContext(createJmsPoolConnection(username, password), sessionMode);
        } catch (JMSException e) {
            throw JMSExceptionSupport.createRuntimeException(e);
        }
    }
}
 

@Test
public void testReserveNextSubscriptionLinkNameSharedVolatileWithNonMatchingTopic() {
    String topicName = "myTopic";
    String subscriptionName1 = "mySubscription1";

    AmqpSubscriptionTracker tracker = new AmqpSubscriptionTracker();

    // For the first shared sub name with Topic
    JmsConsumerInfo sub1consumer1 = createConsumerInfo(subscriptionName1, topicName, true, false, true);
    assertEquals("Unexpected first sub link name", subscriptionName1 + SUB_NAME_DELIMITER + "volatile1", tracker.reserveNextSubscriptionLinkName(subscriptionName1, sub1consumer1));

    // For the next shared sub name with different Topic
    JmsConsumerInfo sub1consumer2 = createConsumerInfo(subscriptionName1, topicName + "-alt", true, false, true);
    try {
        tracker.reserveNextSubscriptionLinkName(subscriptionName1, sub1consumer2);
        fail("Expected JMSRuntimeException when Topic doesn't match previous subscription");
    } catch (JMSRuntimeException jmsre) {
    }
}
 

@Test(timeout = 30000)
public void testCreateJMSProducer() throws JMSException {
    JmsPoolJMSProducer producer = (JmsPoolJMSProducer) context.createProducer();
    assertNotNull(producer);
    MockJMSMessageProducer mockProducer = (MockJMSMessageProducer) producer.getMessageProducer();
    assertNotNull(mockProducer);

    // JMSProducer instances are always anonymous producers.
    assertNull(mockProducer.getDestination());

    context.close();

    try {
        producer.getMessageProducer();
        fail("should throw on closed context.");
    } catch (JMSRuntimeException jmsre) {}
}
 

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

    try {
        session.run();
        fail("Session should be unable to run outside EE.");
    } catch (JMSRuntimeException jmsre) {}

    session.close();

    try {
        session.run();
        fail("Session should be closed.");
    } catch (IllegalStateRuntimeException isre) {}
}
 

@Override
public JMSContext createContext(int sessionMode) {
   switch (sessionMode) {
      case Session.AUTO_ACKNOWLEDGE:
      case Session.CLIENT_ACKNOWLEDGE:
      case Session.DUPS_OK_ACKNOWLEDGE:
      case Session.SESSION_TRANSACTED:
      case ActiveMQJMSConstants.INDIVIDUAL_ACKNOWLEDGE:
      case ActiveMQJMSConstants.PRE_ACKNOWLEDGE:
         break;
      default:
         throw new JMSRuntimeException("Invalid ackmode: " + sessionMode);
   }
   refCounter.increment();

   return new ActiveMQJMSContext(this, sessionMode, threadAwareContext);
}
 

@Test(timeout = 30000)
public void testStartStopConnection() throws JMSException {
    JmsPoolJMSContext context = (JmsPoolJMSContext) cf.createContext();
    context.setAutoStart(false);
    assertNotNull(context.createConsumer(context.createQueue(getTestName())));

    MockJMSConnection connection = (MockJMSConnection) context.getConnection();
    assertFalse(connection.isStarted());

    context.start();
    assertTrue(connection.isStarted());

    // We cannot stop a JMS Connection from the pool as it is a shared resource.
    context.stop();
    assertTrue(connection.isStarted());
    context.close();

    try {
        context.stop();
        fail("Cannot call stop on a closed context.");
    } catch (JMSRuntimeException jmsre) {}
}
 

/**
 * Will receive messages from consumer.  If timeout is hit, consumer.receive(timeout)
 * will return null, and the observable will be completed.
 */
private void receiveLoop() {
  Message message;
  try {
    while ( !closed.get() && ( message = consumer.receive( receiverTimeout ) ) != null ) {
      streamStep.logDebug( message.toString() );
      Date date = new Date( message.getJMSTimestamp() );
      DateFormat formatter = new SimpleDateFormat( "MM-dd-yyyy HH:mm:ss a" );
      formatter.setTimeZone( TimeZone.getTimeZone( "UTC" ) );
      String jmsTimestamp = formatter.format( date );
      acceptRows( singletonList( Arrays.asList( message.getBody( Object.class ), jmsDelegate.destinationName, message.getJMSMessageID(), jmsTimestamp, message.getJMSRedelivered() ) ) );
    }
  } catch ( JMSRuntimeException | JMSException jmsException ) {
    error( jmsException );
  } finally {
    super.close();
    if ( !closed.get() ) {
      close();
      streamStep.logBasic( getString( PKG, "JmsStreamSource.HitReceiveTimeout" ) );
    }
  }
}
 

/**
 * Commits the current transaction.
 *
 * @throws RetriableException Operation failed, but connector should continue to retry.
 * @throws ConnectException   Operation failed and connector should stop.
 */
public void commit() throws ConnectException, RetriableException {
    log.trace("[{}] Entry {}.commit", Thread.currentThread().getId(), this.getClass().getName());

    connectInternal();
    try {
        if (inflight) {
            inflight = false;
        }

        jmsCtxt.commit();
    }
    catch (JMSRuntimeException jmse) {
        log.error("JMS exception {}", jmse);
        throw handleException(jmse);
    }

    log.trace("[{}]  Exit {}.commit", Thread.currentThread().getId(), this.getClass().getName());
}
 

@PreDestroy
private void destroy() {
    if (contexts != null) {
        JMSRuntimeException jre = null;
        for (final JMSContext c : contexts.values()) {
            try {
                c.close();
            } catch (final JMSRuntimeException e) {
                jre = e;
            }
        }
        if (jre != null) {
            throw jre;
        }
    }
}
 

/**
 * Internal method to close the connection.
 */
private void closeInternal() {
    log.trace("[{}] Entry {}.closeInternal", Thread.currentThread().getId(), this.getClass().getName());

    try {
        inflight = false;
        inperil = false;
        connected = false;

        if (jmsCtxt != null) {
            jmsCtxt.close();
        }
    }
    catch (JMSRuntimeException jmse) {
        ;
    }
    finally
    {
        jmsCtxt = null;
        log.debug("Connection to MQ closed");
    }

    log.trace("[{}]  Exit {}.closeInternal", Thread.currentThread().getId(), this.getClass().getName());
}
 

private String getSharedDurableSubLinkName(String subscriptionName, JmsConsumerInfo consumerInfo) {
    JmsDestination topic = consumerInfo.getDestination();
    String selector = consumerInfo.getSelector();

    SubDetails subDetails = null;
    if(sharedDurableSubs.containsKey(subscriptionName)) {
        subDetails = sharedDurableSubs.get(subscriptionName);

        if(subDetails.matches(topic, selector)){
            subDetails.addSubscriber(consumerInfo);
        } else {
            throw new JMSRuntimeException("Subscription details dont match existing subscriber.");
        }
    } else {
        subDetails = new SubDetails(topic, selector, consumerInfo);
    }

    sharedDurableSubs.put(subscriptionName, subDetails);

    int count = subDetails.totalSubscriberCount();

    return getDurableSubscriptionLinkName(subscriptionName, consumerInfo.isExplicitClientID(), count);
}
 

@Test
public void testGetAnotherContextFromIt() {
   JMSContext c2 = context.createContext(Session.DUPS_OK_ACKNOWLEDGE);
   Assert.assertNotNull(c2);
   Assert.assertEquals(Session.DUPS_OK_ACKNOWLEDGE, c2.getSessionMode());
   Message m2 = c2.createMessage();
   Assert.assertNotNull(m2);
   c2.close(); // should close its session, but not its (shared) connection
   try {
      c2.createMessage();
      Assert.fail("session should be closed...");
   } catch (JMSRuntimeException expected) {
      // expected
   }
   Message m1 = context.createMessage();
   Assert.assertNotNull("connection must be open", m1);
}
 

@Test
public void sharedNonDurableSubOnDifferentSelector() throws Exception {
   context = cf.createContext();
   try {
      context.createSharedConsumer(topic1, "mySharedCon", "sel = 'sel1'");
      try {
         context.createSharedConsumer(topic1, "mySharedCon", "sel = 'sel2'");
         fail("expected JMSRuntimeException");
      } catch (JMSRuntimeException jmse) {
         //pass
      } catch (Exception e) {
         fail("threw wrong exception expected JMSRuntimeException got " + e);
      }
   } finally {
      context.close();
   }
}
 

@Test
public void sharedNonDurableSubOnDifferentSelectorSrcFilterNull() throws Exception {
   context = cf.createContext();
   try {
      context.createSharedConsumer(topic1, "mySharedCon");
      try {
         context.createSharedConsumer(topic1, "mySharedCon", "sel = 'sel2'");
         fail("expected JMSRuntimeException");
      } catch (JMSRuntimeException jmse) {
         //pass
      } catch (Exception e) {
         fail("threw wrong exception expected JMSRuntimeException got " + e);
      }
   } finally {
      context.close();
   }
}
 

@Test
public void sharedNonDurableSubOnDifferentSelectorTargetFilterNull() throws Exception {
   context = cf.createContext();
   try {
      context.createSharedConsumer(topic1, "mySharedCon", "sel = 'sel1'");
      try {
         context.createSharedConsumer(topic1, "mySharedCon");
         fail("expected JMSRuntimeException");
      } catch (JMSRuntimeException jmse) {
         //pass
      } catch (Exception e) {
         fail("threw wrong exception expected JMSRuntimeException got " + e);
      }
   } finally {
      context.close();
   }
}
 

@Test
public void sharedDurableSubOnDifferentSelector() throws Exception {
   context = cf.createContext();
   try {
      context.createSharedDurableConsumer(topic1, "mySharedCon", "sel = 'sel1'");
      try {
         context.createSharedDurableConsumer(topic1, "mySharedCon", "sel = 'sel2'");
         fail("expected JMSRuntimeException");
      } catch (JMSRuntimeException jmse) {
         //pass
      } catch (Exception e) {
         fail("threw wrong exception expected JMSRuntimeException got " + e);
      }
   } finally {
      context.close();
   }
}
 

@Test
public void sharedDurableSubOnDifferentSelectorTargetFilterNull() throws Exception {
   context = cf.createContext();
   try {
      context.createSharedDurableConsumer(topic1, "mySharedCon", "sel = 'sel1'");
      try {
         context.createSharedDurableConsumer(topic1, "mySharedCon");
         fail("expected JMSRuntimeException");
      } catch (JMSRuntimeException jmse) {
         //pass
      } catch (Exception e) {
         fail("threw wrong exception expected JMSRuntimeException got " + e);
      }
   } finally {
      context.close();
   }
}
 

@Override
public JMSProducer clearProperties() {
   try {
      stringPropertyNames.clear();
      properties.clear();
   } catch (RuntimeException e) {
      throw new JMSRuntimeException(e.getMessage());
   }
   return this;
}
 

@Test public void should_set_error() {
  jms.after();

  String message;
  try {
    producer.send(jms.queue, "foo");
    throw new AssertionError("expected to throw");
  } catch (JMSRuntimeException e) {
    message = e.getMessage();
  }

  testSpanHandler.takeRemoteSpanWithErrorMessage(PRODUCER, message);
}
 

private String getSharedVolatileSubLinkName(String subscriptionName, JmsConsumerInfo consumerInfo) {
    JmsDestination topic = consumerInfo.getDestination();
    String selector = consumerInfo.getSelector();

    SubDetails subDetails = null;
    if(sharedVolatileSubs.containsKey(subscriptionName)) {
        subDetails = sharedVolatileSubs.get(subscriptionName);

        if(subDetails.matches(topic, selector)){
            subDetails.addSubscriber(consumerInfo);
        } else {
            throw new JMSRuntimeException("Subscription details dont match existing subscriber");
        }
    } else {
        subDetails = new SubDetails(topic, selector, consumerInfo);
    }

    sharedVolatileSubs.put(subscriptionName, subDetails);

    String receiverLinkName = subscriptionName + SUB_NAME_DELIMITER;
    int count = subDetails.totalSubscriberCount();

    if (consumerInfo.isExplicitClientID()) {
        receiverLinkName += "volatile" + count;
    } else {
        receiverLinkName += "global-volatile" + count;
    }

    return receiverLinkName;
}
 

private void validateSessionMode(int mode) {
    switch (mode) {
        case JMSContext.SESSION_TRANSACTED:
        case JMSContext.AUTO_ACKNOWLEDGE:
        case JMSContext.CLIENT_ACKNOWLEDGE:
        case JMSContext.DUPS_OK_ACKNOWLEDGE:
            return;
        default:
            throw new JMSRuntimeException("Invalid Session Mode: " + mode);
    }
}
 

public void checkClientJMSVersionSupport(int requiredMajor, int requiredMinor, boolean runtimeEx) throws JMSException {
    if (jmsMajorVersion >= requiredMajor && jmsMinorVersion >= requiredMinor) {
        return;
    }

    String message = "JMS v" + requiredMajor + "." + requiredMinor + " client feature requested, " +
                     "configured client supports JMS v" + jmsMajorVersion + "." + jmsMinorVersion;

    if (runtimeEx) {
        throw new JMSRuntimeException(message);
    } else {
        throw new JMSException(message);
    }
}
 

@Override
public JMSProducer setDeliveryMode(int deliveryMode) {
    switch (deliveryMode) {
        case DeliveryMode.PERSISTENT:
        case DeliveryMode.NON_PERSISTENT:
            this.deliveryMode = deliveryMode;
            return this;
        default:
            throw new JMSRuntimeException(String.format("Invalid DeliveryMode specified: %d", deliveryMode));
    }
}
 

@Test
public void testCreateContextWithNegativeSessionModeThrowsJMSRE() {
    try {
        context.createContext(-1);
        fail("Should throw JMSRuntimeException");
    } catch (JMSRuntimeException jmsre) {
    }
}
 

@Override
public long getDeliveryTime() {
    Object deliveryTime = getMessageAnnotation(JMS_DELIVERY_TIME);
    if (deliveryTime instanceof Number) {
        return ((Number) deliveryTime).longValue();
    } else if (deliveryTime instanceof Date) {
        return ((Date) deliveryTime).getTime();
    } else if (deliveryTime != null) {
        throw new JMSRuntimeException("Unexpected delivery time annotation type: " + deliveryTime.getClass());
    }

    return syntheticDeliveryTime;
}
 

@Override
public JMSProducer setPriority(int priority) {
    if (priority < 0 || priority > 9) {
        throw new JMSRuntimeException(String.format("Priority value given {%d} is out of range (0..9)", priority));
    }

    this.priority = priority;
    return this;
}
 

@Test
public void testDeliveryDelay() {
    JMSProducer producer = context.createProducer();

    assertEquals(0, producer.getDeliveryDelay());
    try {
        producer.setDeliveryDelay(2000);
        fail("Pool JMSProducer can't modify shared session delay mode.");
    } catch (JMSRuntimeException jmsre) {
    }
}
 

@Override
public JMSProducer setJMSCorrelationIDAsBytes(final byte[] correlationID) {
    if (correlationID == null || correlationID.length == 0) {
        throw new JMSRuntimeException("Please specify a non-zero length byte[]");
    }
    jmsHeaderCorrelationIDAsBytes = Arrays.copyOf(correlationID, correlationID.length);
    return this;
}
 

@Test
public void sendToMdb() throws Exception {
    try (final JMSContext context = cf.createContext()) {
        context.createProducer().send(destination, TEXT);
        assertTrue(Listener.sync());
    } catch (final JMSRuntimeException ex) {
        fail(ex.getMessage());
    }
}
 

@Override
public JMSProducer setPriority(int priority) {
    if (priority < 0 || priority > 9) {
        throw new JMSRuntimeException(String.format("Priority value given {%d} is out of range (0..9)", priority));
    }

    this.priority = priority;
    return this;
}