@Test public void testSubscribe() throws JMSException { setupDefaultMockedConnectionBehaviour(); try { cut.subscribe("bobTheBuilder", null, null); fail("Should have thrown an exception here"); } catch (CougarValidationException anticipated) {} cut.notify(evsd, EventTransportMode.Subscribe); try { cut.subscribe("bobTheBuilder", null, null); fail("Should have thrown an exception here"); } catch (CougarValidationException anticipated) {} cut.setDestinationType(JmsEventTransportImpl.DestinationType.DurableTopic); cut.subscribe(EVENT_NAME, null, inbredExecutionObserver); assertEquals("Should have thrown a CougarFrameworkException as a subscription id must be supplied", CougarFrameworkException.class, this.observedResult.getClass()); observedResult = null; cut.subscribe(EVENT_NAME, new Object[] { "myConnectionId" }, inbredExecutionObserver); assertNotNull(observedResult); assertTrue(observedResult instanceof Subscription); ArgumentCaptor<Monitor> monitorCaptor = ArgumentCaptor.forClass(Monitor.class); verify(monitorRegistry, times(2)).addMonitor(monitorCaptor.capture()); Assert.assertTrue(containsInstanceOf(monitorCaptor.getAllValues(), TopicSubscriberPingMonitor.class)); Assert.assertTrue(containsInstanceOf(monitorCaptor.getAllValues(), ConnectionMonitor.class)); TextMessage mockedMessage = Mockito.mock(TextMessage.class); mockedMessage.setText("{\"name\":\"bob\", \"address\":\"100 chancellors\"}"); messageListener.onMessage(mockedMessage); assertNotNull(observedResult); }
@Override public void subscribe(String eventName, Object[] args, final ExecutionObserver observer) { final Class<? extends Event> eventClass = getEventClass(eventName); String destinationName = destinationResolver.resolveDestination(eventClass, null); String subId = null; if (destinationType == destinationType.DurableTopic) { if (args == null || "".equals(args[0])) { observer.onResult(new ExecutionResult(new CougarFrameworkException("Durable subscription requires a subscription Identifier to be set as the zeroth arg!"))); return; } subId = args[0].toString(); } try { final Session session = getConnection().createSession(false, acknowledgementMode); final TopicSubscriberPingMonitor pingMonitor = setupPingConsumer(eventName, destinationName, subId, session); MessageConsumer consumer = createConsumer(session, destinationName, subId); consumer.setMessageListener(new SubscriptionMessageListener(observer, eventClass, pingMonitor)); subscriptionAdded(observer); observer.onResult(new ExecutionResult(new DefaultSubscription() { @Override public void preClose(CloseReason reason) { try { if (pingMonitor != null) { if (monitorRegistry != null) { monitorRegistry.removeMonitor(pingMonitor); } subscriberMonitorsBySession.remove(pingMonitor); } session.close(); } catch (JMSException ex) { LOGGER.error("Exception occurred when trying to close JMSConnection", ex); } } })); } catch (JMSException ex) { observer.onResult(new ExecutionResult(new CougarFrameworkException("Subscription exception!", ex))); } catch (CougarException ce) { observer.onResult(new ExecutionResult(ce)); } }
JmsEventTransportImpl extends AbstractRegisterableTransport implements EventTransport, ApplicationContextAware, InitializingBean { @Override public void subscribe(String eventName, Object[] args, final ExecutionObserver observer) { final Class<? extends Event> eventClass = getEventClass(eventName); String destinationName = destinationResolver.resolveDestination(eventClass, null); String subId = null; if (destinationType == destinationType.DurableTopic) { if (args == null || "".equals(args[0])) { observer.onResult(new ExecutionResult(new CougarFrameworkException("Durable subscription requires a subscription Identifier to be set as the zeroth arg!"))); return; } subId = args[0].toString(); } try { final Session session = getConnection().createSession(false, acknowledgementMode); final TopicSubscriberPingMonitor pingMonitor = setupPingConsumer(eventName, destinationName, subId, session); MessageConsumer consumer = createConsumer(session, destinationName, subId); consumer.setMessageListener(new SubscriptionMessageListener(observer, eventClass, pingMonitor)); subscriptionAdded(observer); observer.onResult(new ExecutionResult(new DefaultSubscription() { @Override public void preClose(CloseReason reason) { try { if (pingMonitor != null) { if (monitorRegistry != null) { monitorRegistry.removeMonitor(pingMonitor); } subscriberMonitorsBySession.remove(pingMonitor); } session.close(); } catch (JMSException ex) { LOGGER.error("Exception occurred when trying to close JMSConnection", ex); } } })); } catch (JMSException ex) { observer.onResult(new ExecutionResult(new CougarFrameworkException("Subscription exception!", ex))); } catch (CougarException ce) { observer.onResult(new ExecutionResult(ce)); } } }
JmsEventTransportImpl extends AbstractRegisterableTransport implements EventTransport, ApplicationContextAware, InitializingBean { @Override public void subscribe(String eventName, Object[] args, final ExecutionObserver observer) { final Class<? extends Event> eventClass = getEventClass(eventName); String destinationName = destinationResolver.resolveDestination(eventClass, null); String subId = null; if (destinationType == destinationType.DurableTopic) { if (args == null || "".equals(args[0])) { observer.onResult(new ExecutionResult(new CougarFrameworkException("Durable subscription requires a subscription Identifier to be set as the zeroth arg!"))); return; } subId = args[0].toString(); } try { final Session session = getConnection().createSession(false, acknowledgementMode); final TopicSubscriberPingMonitor pingMonitor = setupPingConsumer(eventName, destinationName, subId, session); MessageConsumer consumer = createConsumer(session, destinationName, subId); consumer.setMessageListener(new SubscriptionMessageListener(observer, eventClass, pingMonitor)); subscriptionAdded(observer); observer.onResult(new ExecutionResult(new DefaultSubscription() { @Override public void preClose(CloseReason reason) { try { if (pingMonitor != null) { if (monitorRegistry != null) { monitorRegistry.removeMonitor(pingMonitor); } subscriberMonitorsBySession.remove(pingMonitor); } session.close(); } catch (JMSException ex) { LOGGER.error("Exception occurred when trying to close JMSConnection", ex); } } })); } catch (JMSException ex) { observer.onResult(new ExecutionResult(new CougarFrameworkException("Subscription exception!", ex))); } catch (CougarException ce) { observer.onResult(new ExecutionResult(ce)); } } JmsEventTransportImpl(ConnectionFactory connectionFactory); }
JmsEventTransportImpl extends AbstractRegisterableTransport implements EventTransport, ApplicationContextAware, InitializingBean { @Override public void subscribe(String eventName, Object[] args, final ExecutionObserver observer) { final Class<? extends Event> eventClass = getEventClass(eventName); String destinationName = destinationResolver.resolveDestination(eventClass, null); String subId = null; if (destinationType == destinationType.DurableTopic) { if (args == null || "".equals(args[0])) { observer.onResult(new ExecutionResult(new CougarFrameworkException("Durable subscription requires a subscription Identifier to be set as the zeroth arg!"))); return; } subId = args[0].toString(); } try { final Session session = getConnection().createSession(false, acknowledgementMode); final TopicSubscriberPingMonitor pingMonitor = setupPingConsumer(eventName, destinationName, subId, session); MessageConsumer consumer = createConsumer(session, destinationName, subId); consumer.setMessageListener(new SubscriptionMessageListener(observer, eventClass, pingMonitor)); subscriptionAdded(observer); observer.onResult(new ExecutionResult(new DefaultSubscription() { @Override public void preClose(CloseReason reason) { try { if (pingMonitor != null) { if (monitorRegistry != null) { monitorRegistry.removeMonitor(pingMonitor); } subscriberMonitorsBySession.remove(pingMonitor); } session.close(); } catch (JMSException ex) { LOGGER.error("Exception occurred when trying to close JMSConnection", ex); } } })); } catch (JMSException ex) { observer.onResult(new ExecutionResult(new CougarFrameworkException("Subscription exception!", ex))); } catch (CougarException ce) { observer.onResult(new ExecutionResult(ce)); } } JmsEventTransportImpl(ConnectionFactory connectionFactory); void init(); void initThreadPool(); void initialiseExecutionContext(); void destroy(); @Override void publish(Event event); void publish(Event event, String destinationName, EventServiceBindingDescriptor eventServiceBindingDescriptor); Future<Boolean> requestConnectionToBroker(); void closeConnection(); @Override void subscribe(String eventName, Object[] args, final ExecutionObserver observer); DestinationType getDestinationType(); @Required void setDestinationType(DestinationType destinationType); void setThreadPoolSize(int threadPoolSize); JMSEventMarshaller getEventMarshaller(); @Required void setEventMarshaller(JMSEventMarshaller eventMarshaller); EventUnMarshaller getEventUnMarshaller(); @Required void setEventUnMarshaller(EventUnMarshaller eventUnMarshaller); EventErrorHandler getErrorHandler(); @Required void setErrorHandler(EventErrorHandler errorHandler); @Override void notify(BindingDescriptor bindingDescriptor); @Override void notify(BindingDescriptor bindingDescriptor, EventTransportMode mode); JMSDestinationResolver<String> getDestinationResolver(); @Required void setDestinationResolver(JMSDestinationResolver<String> destinationResolver); void setMonitorRegistry(MonitorRegistry monitorRegistry); @Required void setConnectionMonitor(ConnectionMonitor connectionMonitor); @Override void setApplicationContext(ApplicationContext applicationContext); @Override void afterPropertiesSet(); int getAcknowledgementMode(); void setAcknowledgementMode(int acknowledgementMode); ExecutionContext getExecutionContext(); void setExecutionContext(final ExecutionContext executionContext); String getTransportIdentifier(); void setTransportIdentifier(String transportIdentifier); String getUsername(); @Required void setUsername(String username); String getPassword(); @Required void setPassword(String password); }
JmsEventTransportImpl extends AbstractRegisterableTransport implements EventTransport, ApplicationContextAware, InitializingBean { @Override public void subscribe(String eventName, Object[] args, final ExecutionObserver observer) { final Class<? extends Event> eventClass = getEventClass(eventName); String destinationName = destinationResolver.resolveDestination(eventClass, null); String subId = null; if (destinationType == destinationType.DurableTopic) { if (args == null || "".equals(args[0])) { observer.onResult(new ExecutionResult(new CougarFrameworkException("Durable subscription requires a subscription Identifier to be set as the zeroth arg!"))); return; } subId = args[0].toString(); } try { final Session session = getConnection().createSession(false, acknowledgementMode); final TopicSubscriberPingMonitor pingMonitor = setupPingConsumer(eventName, destinationName, subId, session); MessageConsumer consumer = createConsumer(session, destinationName, subId); consumer.setMessageListener(new SubscriptionMessageListener(observer, eventClass, pingMonitor)); subscriptionAdded(observer); observer.onResult(new ExecutionResult(new DefaultSubscription() { @Override public void preClose(CloseReason reason) { try { if (pingMonitor != null) { if (monitorRegistry != null) { monitorRegistry.removeMonitor(pingMonitor); } subscriberMonitorsBySession.remove(pingMonitor); } session.close(); } catch (JMSException ex) { LOGGER.error("Exception occurred when trying to close JMSConnection", ex); } } })); } catch (JMSException ex) { observer.onResult(new ExecutionResult(new CougarFrameworkException("Subscription exception!", ex))); } catch (CougarException ce) { observer.onResult(new ExecutionResult(ce)); } } JmsEventTransportImpl(ConnectionFactory connectionFactory); void init(); void initThreadPool(); void initialiseExecutionContext(); void destroy(); @Override void publish(Event event); void publish(Event event, String destinationName, EventServiceBindingDescriptor eventServiceBindingDescriptor); Future<Boolean> requestConnectionToBroker(); void closeConnection(); @Override void subscribe(String eventName, Object[] args, final ExecutionObserver observer); DestinationType getDestinationType(); @Required void setDestinationType(DestinationType destinationType); void setThreadPoolSize(int threadPoolSize); JMSEventMarshaller getEventMarshaller(); @Required void setEventMarshaller(JMSEventMarshaller eventMarshaller); EventUnMarshaller getEventUnMarshaller(); @Required void setEventUnMarshaller(EventUnMarshaller eventUnMarshaller); EventErrorHandler getErrorHandler(); @Required void setErrorHandler(EventErrorHandler errorHandler); @Override void notify(BindingDescriptor bindingDescriptor); @Override void notify(BindingDescriptor bindingDescriptor, EventTransportMode mode); JMSDestinationResolver<String> getDestinationResolver(); @Required void setDestinationResolver(JMSDestinationResolver<String> destinationResolver); void setMonitorRegistry(MonitorRegistry monitorRegistry); @Required void setConnectionMonitor(ConnectionMonitor connectionMonitor); @Override void setApplicationContext(ApplicationContext applicationContext); @Override void afterPropertiesSet(); int getAcknowledgementMode(); void setAcknowledgementMode(int acknowledgementMode); ExecutionContext getExecutionContext(); void setExecutionContext(final ExecutionContext executionContext); String getTransportIdentifier(); void setTransportIdentifier(String transportIdentifier); String getUsername(); @Required void setUsername(String username); String getPassword(); @Required void setPassword(String password);  }