ArtemisMessageSource implements MessageSource { @Override public void onStatusUpdate(QpidStatus oldStatus, QpidStatus newStatus) { if (newStatus.equals(oldStatus)) { return; } log.debug("ArtemisMessageSource was notified of a QpidStatus change: {}", newStatus); for (MessageReceiver receiver : this.messageReceivers) { if (!receiver.requiresQpid()) { continue; } switch (newStatus) { case FLOW_STOPPED: case MISSING_BINDING: case MISSING_EXCHANGE: case DOWN: receiver.pause(); break; case CONNECTED: receiver.resume(); break; default: } } } @Inject  ArtemisMessageSource(ActiveMQSessionFactory sessionFactory, ObjectMapper mapper,
        MessageSourceReceiverFactory receiverFactory); @Override void shutDown(); @Override void onStatusUpdate(QpidStatus oldStatus, QpidStatus newStatus); @Override void onStatusUpdate(ActiveMQStatus oldStatus, ActiveMQStatus newStatus); static String getQueueName(EventListener listener);  }
@Test public void eventSourceDoesNotConnectEventReceiversUntilNotifiedConnectionEstablished() throws Exception { ActiveMQSessionFactory connection = createConnection(); List<MessageReceiver> messageReceivers = new LinkedList<>(); messageReceivers.add(new DefaultEventMessageReceiver(mock(EventListener.class), connection, mock(ObjectMapper.class))); MessageSourceReceiverFactory receiverFactory = mock(MessageSourceReceiverFactory.class); when(receiverFactory.get(eq(connection))).thenReturn(messageReceivers); ArtemisMessageSource source = new ArtemisMessageSource(connection, new ObjectMapper(), receiverFactory); verifyZeroInteractions(clientSessionFactory); ClientSession session = mock(ClientSession.class); when(clientSessionFactory.createSession()).thenReturn(session); when(session.createConsumer(any(String.class))).thenReturn(mock(ClientConsumer.class)); source.onStatusUpdate(ActiveMQStatus.UNKNOWN, ActiveMQStatus.CONNECTED); verify(session).createConsumer(any(String.class)); verify(session, times(1)).start(); verify(session, never()).stop(); }
@Test public void eventSourceDoesNothingWhenQpidStatusDoesntChange() throws Exception { ClientSession session = mock(ClientSession.class); when(clientSessionFactory.createSession()).thenReturn(session); when(session.createConsumer(any(String.class))).thenReturn(mock(ClientConsumer.class)); ArtemisMessageSource messageSource = createEventSourceStubbedWithFactoryCreation(createQpidReceiver()); messageSource.onStatusUpdate(QpidStatus.DOWN, QpidStatus.DOWN); verify(session, times(1)).start(); verify(session, never()).stop(); }
@Test public void eventSourceClosesEventReceiverClientConsumerWhenQpidGoesDown() throws Exception { ClientSession session = mock(ClientSession.class); when(clientSessionFactory.createSession()).thenReturn(session); ClientConsumer consumer = mock(ClientConsumer.class); when(session.createConsumer(any(String.class))).thenReturn(consumer); ArtemisMessageSource messageSource = createEventSourceStubbedWithFactoryCreation(createQpidReceiver()); messageSource.onStatusUpdate(QpidStatus.CONNECTED, QpidStatus.DOWN); verify(session, times(1)).start(); verify(consumer).close(); verify(session, never()).stop(); }
@Test public void eventSourceCreatesNewEventReceiverClientSessionWhenQpidComesUp() throws Exception { ClientSession session = mock(ClientSession.class); when(clientSessionFactory.createSession()).thenReturn(session); ClientConsumer consumer1 = mock(ClientConsumer.class); when(consumer1.isClosed()).thenReturn(true); ClientConsumer consumer2 = mock(ClientConsumer.class); when(session.createConsumer(any(String.class))).thenReturn(consumer1, consumer2); QpidEventMessageReceiver receiver = createQpidReceiver(); ArtemisMessageSource messageSource = createEventSourceStubbedWithFactoryCreation(receiver); messageSource.onStatusUpdate(QpidStatus.DOWN, QpidStatus.CONNECTED); verify(session, times(1)).start(); verify(session, never()).stop(); verify(session, times(2)).createConsumer(any(String.class)); }
@Test public void eventSourceDoesNothingWithEventReceiverClientSessionWhenListenerDoesntRequireQpid() throws Exception { ClientSession session = mock(ClientSession.class); when(clientSessionFactory.createSession()).thenReturn(session); when(session.createConsumer(any(String.class))).thenReturn(mock(ClientConsumer.class)); DefaultEventMessageReceiver receiver = new DefaultEventMessageReceiver(mock(EventListener.class), sessionFactory, mock(ObjectMapper.class)); ArtemisMessageSource messageSource = createEventSourceStubbedWithFactoryCreation(receiver); messageSource.onStatusUpdate(QpidStatus.DOWN, QpidStatus.CONNECTED); verify(session, times(1)).start(); verify(session, never()).stop(); }