@Test public void testDurableTopicConsumer() throws Exception { JmsDestination jmsDestination = mockTopicJmsDestination(false); mockCreateDestinationAndConsumer(jmsDestination, RFC6167_SUBSCRIPTION); TopicSubscriber topicSubscriber = mock(TopicSubscriber.class); when(session.createDurableSubscriber(any(Topic.class), eq(SUBSCRIPTION_ID))).thenReturn(topicSubscriber); JmsMessageConsumerFactory jmsMessageConsumerFactory = new JmsMessageConsumerFactory(mockVendor, session, RFC6167_SUBSCRIPTION, false, FILTER, jmsActorConfig); MessageConsumer messageConsumer = jmsMessageConsumerFactory.create(); assertEquals(topicSubscriber, messageConsumer); verify(session).createDurableSubscriber(any(Topic.class), eq(SUBSCRIPTION_ID)); }
public MessageConsumer create() throws JMSException { JmsDestination destination = vendor.createDestination(rfc6167, jmsActorConfig); MessageConsumer consumer = null; if (deferConsumerCreationToVendor) { consumer = vendor.createConsumer(destination, filterExp, jmsActorConfig); } else { if (destination.destinationType().equals(DestinationType.TOPIC)) { if (!isEmpty(destination.subscriptionId())) { if (!isEmpty(destination.sharedConsumerId())) { log.trace("Creating new shared durable consumer."); consumer = ((ConsumerCreator) (session, dest, filterExpression) -> session .createSharedDurableConsumer((Topic) dest.getDestination(), dest.subscriptionId(), filterExpression)) .createConsumer(session, destination, filterExp); } else { log.trace("Creating new durable consumer."); consumer = ((ConsumerCreator) (session, dest, filterExpression) -> session .createDurableSubscriber((Topic) dest.getDestination(), dest.subscriptionId())).createConsumer(session, destination, filterExp); } } else if (!isEmpty(destination.sharedConsumerId())) { log.trace("Creating new shared consumer."); consumer = ((ConsumerCreator) (session, dest, filterExpression) -> session .createSharedConsumer((Topic) dest.getDestination(), dest.sharedConsumerId(), filterExpression)).createConsumer(session, destination, filterExp); } } if (consumer == null) { log.trace("Creating new standard consumer."); consumer = ((ConsumerCreator) (session, dest, filterExpression) -> session.createConsumer(dest.getDestination(), filterExpression)) .createConsumer(session, destination, filterExp); } } return consumer; }
JmsMessageConsumerFactory { public MessageConsumer create() throws JMSException { JmsDestination destination = vendor.createDestination(rfc6167, jmsActorConfig); MessageConsumer consumer = null; if (deferConsumerCreationToVendor) { consumer = vendor.createConsumer(destination, filterExp, jmsActorConfig); } else { if (destination.destinationType().equals(DestinationType.TOPIC)) { if (!isEmpty(destination.subscriptionId())) { if (!isEmpty(destination.sharedConsumerId())) { log.trace("Creating new shared durable consumer."); consumer = ((ConsumerCreator) (session, dest, filterExpression) -> session .createSharedDurableConsumer((Topic) dest.getDestination(), dest.subscriptionId(), filterExpression)) .createConsumer(session, destination, filterExp); } else { log.trace("Creating new durable consumer."); consumer = ((ConsumerCreator) (session, dest, filterExpression) -> session .createDurableSubscriber((Topic) dest.getDestination(), dest.subscriptionId())).createConsumer(session, destination, filterExp); } } else if (!isEmpty(destination.sharedConsumerId())) { log.trace("Creating new shared consumer."); consumer = ((ConsumerCreator) (session, dest, filterExpression) -> session .createSharedConsumer((Topic) dest.getDestination(), dest.sharedConsumerId(), filterExpression)).createConsumer(session, destination, filterExp); } } if (consumer == null) { log.trace("Creating new standard consumer."); consumer = ((ConsumerCreator) (session, dest, filterExpression) -> session.createConsumer(dest.getDestination(), filterExpression)) .createConsumer(session, destination, filterExp); } } return consumer; } }
JmsMessageConsumerFactory { public MessageConsumer create() throws JMSException { JmsDestination destination = vendor.createDestination(rfc6167, jmsActorConfig); MessageConsumer consumer = null; if (deferConsumerCreationToVendor) { consumer = vendor.createConsumer(destination, filterExp, jmsActorConfig); } else { if (destination.destinationType().equals(DestinationType.TOPIC)) { if (!isEmpty(destination.subscriptionId())) { if (!isEmpty(destination.sharedConsumerId())) { log.trace("Creating new shared durable consumer."); consumer = ((ConsumerCreator) (session, dest, filterExpression) -> session .createSharedDurableConsumer((Topic) dest.getDestination(), dest.subscriptionId(), filterExpression)) .createConsumer(session, destination, filterExp); } else { log.trace("Creating new durable consumer."); consumer = ((ConsumerCreator) (session, dest, filterExpression) -> session .createDurableSubscriber((Topic) dest.getDestination(), dest.subscriptionId())).createConsumer(session, destination, filterExp); } } else if (!isEmpty(destination.sharedConsumerId())) { log.trace("Creating new shared consumer."); consumer = ((ConsumerCreator) (session, dest, filterExpression) -> session .createSharedConsumer((Topic) dest.getDestination(), dest.sharedConsumerId(), filterExpression)).createConsumer(session, destination, filterExp); } } if (consumer == null) { log.trace("Creating new standard consumer."); consumer = ((ConsumerCreator) (session, dest, filterExpression) -> session.createConsumer(dest.getDestination(), filterExpression)) .createConsumer(session, destination, filterExp); } } return consumer; } JmsMessageConsumerFactory(VendorImplementation vendor, Session session, String rfc6167, boolean deferConsumerCreationToVendor,
      String filterExp, JmsActorConfig jmsActorConfig); }
JmsMessageConsumerFactory { public MessageConsumer create() throws JMSException { JmsDestination destination = vendor.createDestination(rfc6167, jmsActorConfig); MessageConsumer consumer = null; if (deferConsumerCreationToVendor) { consumer = vendor.createConsumer(destination, filterExp, jmsActorConfig); } else { if (destination.destinationType().equals(DestinationType.TOPIC)) { if (!isEmpty(destination.subscriptionId())) { if (!isEmpty(destination.sharedConsumerId())) { log.trace("Creating new shared durable consumer."); consumer = ((ConsumerCreator) (session, dest, filterExpression) -> session .createSharedDurableConsumer((Topic) dest.getDestination(), dest.subscriptionId(), filterExpression)) .createConsumer(session, destination, filterExp); } else { log.trace("Creating new durable consumer."); consumer = ((ConsumerCreator) (session, dest, filterExpression) -> session .createDurableSubscriber((Topic) dest.getDestination(), dest.subscriptionId())).createConsumer(session, destination, filterExp); } } else if (!isEmpty(destination.sharedConsumerId())) { log.trace("Creating new shared consumer."); consumer = ((ConsumerCreator) (session, dest, filterExpression) -> session .createSharedConsumer((Topic) dest.getDestination(), dest.sharedConsumerId(), filterExpression)).createConsumer(session, destination, filterExp); } } if (consumer == null) { log.trace("Creating new standard consumer."); consumer = ((ConsumerCreator) (session, dest, filterExpression) -> session.createConsumer(dest.getDestination(), filterExpression)) .createConsumer(session, destination, filterExp); } } return consumer; } JmsMessageConsumerFactory(VendorImplementation vendor, Session session, String rfc6167, boolean deferConsumerCreationToVendor,
      String filterExp, JmsActorConfig jmsActorConfig); MessageConsumer create(); }
JmsMessageConsumerFactory { public MessageConsumer create() throws JMSException { JmsDestination destination = vendor.createDestination(rfc6167, jmsActorConfig); MessageConsumer consumer = null; if (deferConsumerCreationToVendor) { consumer = vendor.createConsumer(destination, filterExp, jmsActorConfig); } else { if (destination.destinationType().equals(DestinationType.TOPIC)) { if (!isEmpty(destination.subscriptionId())) { if (!isEmpty(destination.sharedConsumerId())) { log.trace("Creating new shared durable consumer."); consumer = ((ConsumerCreator) (session, dest, filterExpression) -> session .createSharedDurableConsumer((Topic) dest.getDestination(), dest.subscriptionId(), filterExpression)) .createConsumer(session, destination, filterExp); } else { log.trace("Creating new durable consumer."); consumer = ((ConsumerCreator) (session, dest, filterExpression) -> session .createDurableSubscriber((Topic) dest.getDestination(), dest.subscriptionId())).createConsumer(session, destination, filterExp); } } else if (!isEmpty(destination.sharedConsumerId())) { log.trace("Creating new shared consumer."); consumer = ((ConsumerCreator) (session, dest, filterExpression) -> session .createSharedConsumer((Topic) dest.getDestination(), dest.sharedConsumerId(), filterExpression)).createConsumer(session, destination, filterExp); } } if (consumer == null) { log.trace("Creating new standard consumer."); consumer = ((ConsumerCreator) (session, dest, filterExpression) -> session.createConsumer(dest.getDestination(), filterExpression)) .createConsumer(session, destination, filterExp); } } return consumer; } JmsMessageConsumerFactory(VendorImplementation vendor, Session session, String rfc6167, boolean deferConsumerCreationToVendor,
      String filterExp, JmsActorConfig jmsActorConfig); MessageConsumer create();  }