StandardWorkflow extends StandardWorkflowImpl { @Override public synchronized void onAdaptrisMessage(AdaptrisMessage msg, Consumer<AdaptrisMessage> success, Consumer<AdaptrisMessage> failure) { super.onAdaptrisMessage(msg, success, failure); } StandardWorkflow();  StandardWorkflow(AdaptrisMessageConsumer consumer, AdaptrisMessageProducer producer);  StandardWorkflow(AdaptrisMessageConsumer consumer, ServiceCollection services, AdaptrisMessageProducer producer); @Override synchronized void onAdaptrisMessage(AdaptrisMessage msg, Consumer<AdaptrisMessage> success, Consumer<AdaptrisMessage> failure);  }
@Test public void testOnMessageWithSendEvents() throws Exception { MockMessageProducer producer = new MockMessageProducer(); MockMessageProducer eventProd = new MockMessageProducer(); DefaultEventHandler evtHandler = new DefaultEventHandler(eventProd); MockChannel channel = createChannel(producer, Arrays.asList(new Service[] { new AddMetadataService(Arrays.asList(new MetadataElement[] { new MetadataElement(METADATA_KEY, METADATA_VALUE) })), new PayloadFromMetadataService(PAYLOAD_2) })); try { channel.setEventHandler(evtHandler); StandardWorkflow workflow = (StandardWorkflow) channel.getWorkflowList().get(0); workflow.setSendEvents(true); channel.prepare(); AdaptrisMessage msg = AdaptrisMessageFactory.getDefaultInstance().newMessage(PAYLOAD_1); start(channel); workflow.onAdaptrisMessage(msg); assertEquals("Make sure all produced", 1, producer.getMessages().size()); for (Iterator i = producer.getMessages().iterator(); i.hasNext();) { AdaptrisMessage m = (AdaptrisMessage) i.next(); assertEquals(PAYLOAD_2, m.getContent()); assertTrue("Contains correct metadata key", m.containsKey(METADATA_KEY)); assertEquals(METADATA_VALUE, m.getMetadataValue(METADATA_KEY)); } waitForMessages(eventProd, 1); assertEquals(1, eventProd.messageCount()); } finally { stop(channel); stop(evtHandler); } }
@Test public void testOnMessageWithoutEvents() throws Exception { MockMessageProducer producer = new MockMessageProducer(); MockMessageProducer eventProd = new MockMessageProducer(); DefaultEventHandler evtHandler = new DefaultEventHandler(eventProd); MockChannel channel = createChannel(producer, Arrays.asList(new Service[] { new AddMetadataService(Arrays.asList(new MetadataElement[] { new MetadataElement(METADATA_KEY, METADATA_VALUE) })), new PayloadFromMetadataService(PAYLOAD_2) })); try { channel.setEventHandler(evtHandler); StandardWorkflow workflow = (StandardWorkflow) channel.getWorkflowList().get(0); workflow.setSendEvents(false); channel.prepare(); AdaptrisMessage msg = AdaptrisMessageFactory.getDefaultInstance().newMessage(PAYLOAD_1); start(channel); workflow.onAdaptrisMessage(msg); assertEquals("Make sure all produced", 1, producer.getMessages().size()); for (Iterator i = producer.getMessages().iterator(); i.hasNext();) { AdaptrisMessage m = (AdaptrisMessage) i.next(); assertEquals(PAYLOAD_2, m.getContent()); assertTrue("Contains correct metadata key", m.containsKey(METADATA_KEY)); assertEquals(METADATA_VALUE, m.getMetadataValue(METADATA_KEY)); } waitForMessages(eventProd, 0); assertEquals(0, eventProd.messageCount()); } finally { stop(channel); stop(evtHandler); } }
@Test public void testOnMessageWithInterceptors() throws Exception { MockWorkflowInterceptor interceptor = new MockWorkflowInterceptor(); MockMessageProducer producer = new MockMessageProducer(); MockChannel channel = createChannel(producer, Arrays.asList(new Service[] { new AddMetadataService(Arrays.asList(new MetadataElement[] { new MetadataElement(METADATA_KEY, METADATA_VALUE) })), new PayloadFromMetadataService(PAYLOAD_2) })); int count = 10; try { StandardWorkflow workflow = (StandardWorkflow) channel.getWorkflowList().get(0); workflow.addInterceptor(interceptor); channel.prepare(); start(channel); for (int i = 0; i < count; i++) { AdaptrisMessage msg = AdaptrisMessageFactory.getDefaultInstance().newMessage(PAYLOAD_1); workflow.onAdaptrisMessage(msg); } assertEquals("Make sure all produced", count, producer.messageCount()); assertEquals("Make sure all intercepted", count, interceptor.messageCount()); for (AdaptrisMessage m : producer.getMessages()) { assertEquals(PAYLOAD_2, m.getContent()); assertTrue("Contains correct metadata key", m.containsKey(METADATA_KEY)); assertEquals(METADATA_VALUE, m.getMetadataValue(METADATA_KEY)); } } finally { stop(channel); } }
@Test public void testHandleChannelUnavailable() throws Exception { MockMessageProducer producer = new MockMessageProducer(); final MockChannel channel = createChannel(producer, Arrays.asList(new Service[] { new AddMetadataService(Arrays.asList(new MetadataElement[] { new MetadataElement(METADATA_KEY, METADATA_VALUE) })), new PayloadFromMetadataService(PAYLOAD_2) })); try { StandardWorkflow workflow = (StandardWorkflow) channel.getWorkflowList().get(0); workflow.setChannelUnavailableWaitInterval(new TimeInterval(1200L, TimeUnit.MILLISECONDS)); AdaptrisMessage msg = AdaptrisMessageFactory.getDefaultInstance().newMessage(PAYLOAD_1); channel.prepare(); start(channel); channel.toggleAvailability(false); Timer t = new Timer(true); t.schedule(new TimerTask() { @Override public void run() { channel.toggleAvailability(true); } }, 500); workflow.onAdaptrisMessage(msg); assertEquals("Make sure all produced", 1, producer.getMessages().size()); for (Iterator i = producer.getMessages().iterator(); i.hasNext();) { AdaptrisMessage m = (AdaptrisMessage) i.next(); assertEquals(PAYLOAD_2, m.getContent()); assertTrue("Contains correct metadata key", m.containsKey(METADATA_KEY)); assertEquals(METADATA_VALUE, m.getMetadataValue(METADATA_KEY)); } } finally { stop(channel); } }
@Test public void testHandleChannelUnavailableForever() throws Exception { MockMessageProducer producer = new MockMessageProducer(); final MockChannel channel = createChannel(producer, Arrays.asList(new Service[] { new AddMetadataService(Arrays.asList(new MetadataElement[] { new MetadataElement(METADATA_KEY, METADATA_VALUE) })), new PayloadFromMetadataService(PAYLOAD_2) })); MockMessageProducer meh = new MockMessageProducer(); try { StandardWorkflow workflow = (StandardWorkflow) channel.getWorkflowList().get(0); workflow.setMessageErrorHandler(new StandardProcessingExceptionHandler( new ServiceList(new ArrayList<Service>(Arrays.asList(new Service[] { new StandaloneProducer(meh) }))))); workflow.setChannelUnavailableWaitInterval(new TimeInterval(100L, TimeUnit.MILLISECONDS)); AdaptrisMessage msg = AdaptrisMessageFactory.getDefaultInstance().newMessage(PAYLOAD_1); channel.prepare(); start(channel); channel.toggleAvailability(false); workflow.onAdaptrisMessage(msg); assertEquals("Make none produced", 0, producer.getMessages().size()); assertEquals(1, meh.getMessages().size()); for (Iterator i = meh.getMessages().iterator(); i.hasNext();) { AdaptrisMessage m = (AdaptrisMessage) i.next(); assertEquals(PAYLOAD_1, m.getContent()); assertFalse("Does not contains correct metadata key", m.containsKey(METADATA_KEY)); } } finally { stop(channel); } }
@Test public void testServiceException() throws Exception { MockMessageProducer producer = new MockMessageProducer(); MockMessageProducer meh = new MockMessageProducer(); MockChannel channel = createChannel(producer, Arrays.asList(new Service[] { new ThrowExceptionService(new ConfiguredException("Fail")) })); try { StandardWorkflow workflow = (StandardWorkflow) channel.getWorkflowList().get(0); channel.setMessageErrorHandler(new StandardProcessingExceptionHandler( new ServiceList(new ArrayList<Service>(Arrays.asList(new Service[] { new StandaloneProducer(meh) }))))); channel.prepare(); AdaptrisMessage msg = AdaptrisMessageFactory.getDefaultInstance().newMessage(PAYLOAD_1); start(channel); workflow.onAdaptrisMessage(msg); assertEquals("Make none produced", 0, producer.getMessages().size()); assertEquals(1, meh.getMessages().size()); for (Iterator i = meh.getMessages().iterator(); i.hasNext();) { AdaptrisMessage m = (AdaptrisMessage) i.next(); assertEquals(PAYLOAD_1, m.getContent()); assertFalse("Does not contains correct metadata key", m.containsKey(METADATA_KEY)); assertNotNull(m.getObjectHeaders().get(CoreConstants.OBJ_METADATA_EXCEPTION)); assertNotNull(m.getObjectHeaders().get(CoreConstants.OBJ_METADATA_EXCEPTION_CAUSE)); assertEquals(ThrowExceptionService.class.getSimpleName(), m.getObjectHeaders().get(CoreConstants.OBJ_METADATA_EXCEPTION_CAUSE)); } } finally { stop(channel); } }
@Test public void testProduceException() throws Exception { MockMessageProducer producer = new MockMessageProducer() { @Override protected void doProduce(AdaptrisMessage msg, String endpoint) throws ProduceException { throw new ProduceException(); } }; MockMessageProducer meh = new MockMessageProducer(); MockChannel channel = createChannel(producer, Arrays.asList(new Service[] { new ThrowExceptionService(new ConfiguredException("Fail")) })); try { StandardWorkflow workflow = (StandardWorkflow) channel.getWorkflowList().get(0); channel.setMessageErrorHandler(new StandardProcessingExceptionHandler( new ServiceList(new ArrayList<Service>(Arrays.asList(new Service[] { new StandaloneProducer(meh) }))))); channel.prepare(); AdaptrisMessage msg = AdaptrisMessageFactory.getDefaultInstance().newMessage(PAYLOAD_1); start(channel); workflow.onAdaptrisMessage(msg); assertEquals("Make none produced", 0, producer.getMessages().size()); assertEquals(1, meh.getMessages().size()); for (Iterator i = meh.getMessages().iterator(); i.hasNext();) { AdaptrisMessage m = (AdaptrisMessage) i.next(); assertEquals(PAYLOAD_1, m.getContent()); assertFalse("Does not contains correct metadata key", m.containsKey(METADATA_KEY)); } } finally { stop(channel); } }
@Test public void testRuntimeException() throws Exception { MockMessageProducer producer = new MockMessageProducer() { @Override protected void doProduce(AdaptrisMessage msg, String endpoint) throws ProduceException { throw new RuntimeException(); } }; MockMessageProducer meh = new MockMessageProducer(); MockChannel channel = createChannel(producer, Arrays.asList(new Service[] { new AddMetadataService(Arrays.asList(new MetadataElement[] { new MetadataElement(METADATA_KEY, METADATA_VALUE) })), new PayloadFromMetadataService(PAYLOAD_2) })); try { StandardWorkflow workflow = (StandardWorkflow) channel.getWorkflowList().get(0); channel.setMessageErrorHandler(new StandardProcessingExceptionHandler( new ServiceList(new ArrayList<Service>(Arrays.asList(new Service[] { new StandaloneProducer(meh) }))))); channel.prepare(); AdaptrisMessage msg = AdaptrisMessageFactory.getDefaultInstance().newMessage(PAYLOAD_1); start(channel); workflow.onAdaptrisMessage(msg); assertEquals("Make none produced", 0, producer.getMessages().size()); assertEquals(1, meh.getMessages().size()); for (Iterator i = meh.getMessages().iterator(); i.hasNext();) { AdaptrisMessage m = (AdaptrisMessage) i.next(); assertEquals(PAYLOAD_1, m.getContent()); assertFalse("Does not contains correct metadata key", m.containsKey(METADATA_KEY)); } } finally { stop(channel); } }
@Test public void testOnMessage_SkipProducer() throws Exception { MockMessageProducer producer = new MockMessageProducer(); MockMessageProducer serviceProducer = new MockMessageProducer(); MockChannel channel = createChannel(producer, Arrays.asList( new Service[] {new StandaloneProducer(serviceProducer), new MockSkipProducerService()})); AdaptrisMessage msg = AdaptrisMessageFactory.getDefaultInstance().newMessage(PAYLOAD_1); StandardWorkflow workflow = (StandardWorkflow) channel.getWorkflowList().get(0); try { start(channel); workflow.onAdaptrisMessage(msg); assertEquals(1, serviceProducer.messageCount()); assertEquals(0, producer.messageCount()); } finally { stop(channel); } }
@Test public void testOnMessage_MessageLogger() throws Exception { MockMessageProducer producer = new MockMessageProducer(); MockChannel channel = createChannel(producer, Arrays.asList(new Service[] {new NullService()})); AdaptrisMessage msg = AdaptrisMessageFactory.getDefaultInstance().newMessage(PAYLOAD_1); StandardWorkflow workflow = (StandardWorkflow) channel.getWorkflowList().get(0); workflow.setMessageLogger(new MinimalMessageLogger()); try { start(channel); workflow.onAdaptrisMessage(msg); assertEquals(1, producer.messageCount()); } finally { stop(channel); } }
@Test public void testOnMessage_withConsumeLocation() throws Exception { MockMessageProducer producer = new MockMessageProducer(); MockChannel channel = createChannel(producer, Arrays.asList(new Service[] {new NullService()})); AdaptrisMessage msg = AdaptrisMessageFactory.getDefaultInstance().newMessage(PAYLOAD_1); msg.addMessageHeader(getName(), "hello world"); StandardWorkflow workflow = (StandardWorkflow) channel.getWorkflowList().get(0); workflow.setConsumer(new ConsumerWithLocation(getName())); try { start(channel); workflow.onAdaptrisMessage(msg); AdaptrisMessage consumed = producer.getMessages().get(0); assertTrue(consumed.headersContainsKey(CoreConstants.MESSAGE_CONSUME_LOCATION)); assertEquals("hello world", consumed.getMetadataValue(CoreConstants.MESSAGE_CONSUME_LOCATION)); } finally { stop(channel); } }
@Test public void testOnMessage_withConsumeLocation_NoMatch() throws Exception { MockMessageProducer producer = new MockMessageProducer(); MockChannel channel = createChannel(producer, Arrays.asList(new Service[] {new NullService()})); AdaptrisMessage msg = AdaptrisMessageFactory.getDefaultInstance().newMessage(PAYLOAD_1); StandardWorkflow workflow = (StandardWorkflow) channel.getWorkflowList().get(0); workflow.setConsumer(new ConsumerWithLocation(getName())); try { start(channel); workflow.onAdaptrisMessage(msg); AdaptrisMessage consumed = producer.getMessages().get(0); assertFalse(consumed.headersContainsKey(CoreConstants.MESSAGE_CONSUME_LOCATION)); } finally { stop(channel); } }
@Test public void testOnMessage_SuccessCallback() throws Exception { AtomicBoolean onSuccess = new AtomicBoolean(false); MockChannel channel = createChannel(new MockMessageProducer(), Arrays.asList(new Service[] {new NullService()})); AdaptrisMessage msg = AdaptrisMessageFactory.getDefaultInstance().newMessage(PAYLOAD_1); StandardWorkflow workflow = (StandardWorkflow) channel.getWorkflowList().get(0); try { start(channel); workflow.onAdaptrisMessage(msg, (m) -> { onSuccess.set(true); }); assertTrue(onSuccess.get()); } finally { stop(channel); } }
@Test public void testOnMessage_ServiceFailsFailureCallback() throws Exception { AtomicBoolean onSuccess = new AtomicBoolean(false); MockChannel channel = createChannel(new MockMessageProducer(), Arrays.asList(new Service[] {new ThrowExceptionService(new ConfiguredException("expected"))})); AdaptrisMessage msg = AdaptrisMessageFactory.getDefaultInstance().newMessage(PAYLOAD_1); StandardWorkflow workflow = (StandardWorkflow) channel.getWorkflowList().get(0); try { start(channel); workflow.onAdaptrisMessage(msg, (m) -> { onSuccess.set(true); }, (m) -> { onSuccess.set(false); }); assertFalse(onSuccess.get()); } finally { stop(channel); } }
@Test public void testOnMessage_ServiceFailsWithRetryHandlerFailureCallback() throws Exception { AtomicBoolean onSuccess = new AtomicBoolean(false); MockChannel channel = createChannel(new MockMessageProducer(), Arrays.asList(new Service[] {new ThrowExceptionService(new ConfiguredException("expected"))})); channel.setMessageErrorHandler(new RetryMessageErrorHandler(3, new TimeInterval(100l, TimeUnit.MILLISECONDS), new NullService())); AdaptrisMessage msg = AdaptrisMessageFactory.getDefaultInstance().newMessage(PAYLOAD_1); StandardWorkflow workflow = (StandardWorkflow) channel.getWorkflowList().get(0); try { start(channel); workflow.onAdaptrisMessage(msg, (m) -> { onSuccess.set(true); }, (m) -> { onSuccess.set(false); }); assertFalse(onSuccess.get()); } finally { stop(channel); } }
@Test public void testOnMessage_ProducerFailsFailureCallback() throws Exception { AtomicBoolean onSuccess = new AtomicBoolean(false); MockChannel channel = createChannel(new FailFirstMockMessageProducer(1), Arrays.asList(new Service[] {new NullService()})); AdaptrisMessage msg = AdaptrisMessageFactory.getDefaultInstance().newMessage(PAYLOAD_1); StandardWorkflow workflow = (StandardWorkflow) channel.getWorkflowList().get(0); try { start(channel); workflow.onAdaptrisMessage(msg, (m) -> { onSuccess.set(true); }, (m) -> { onSuccess.set(false); }); assertFalse(onSuccess.get()); } finally { stop(channel); } }
@Test public void testOnMessage_ProducerFailsWithRetryHandlerSuccessCallback() throws Exception { AtomicBoolean onSuccess = new AtomicBoolean(false); MockChannel channel = createChannel(new FailFirstMockMessageProducer(1), Arrays.asList(new Service[] {new NullService()})); channel.setMessageErrorHandler(new RetryMessageErrorHandler(2, new TimeInterval(100l, TimeUnit.MILLISECONDS), new NullService())); AdaptrisMessage msg = AdaptrisMessageFactory.getDefaultInstance().newMessage(PAYLOAD_1); StandardWorkflow workflow = (StandardWorkflow) channel.getWorkflowList().get(0); try { start(channel); workflow.onAdaptrisMessage(msg, (m) -> { onSuccess.set(true); }, (m) -> { onSuccess.set(false); }); Awaitility.await() .atMost(Duration.ofSeconds(2)) .with() .pollInterval(Duration.ofMillis(100)) .until(onSuccess::get); assertTrue(onSuccess.get()); } finally { stop(channel); } }
@Test public void testOnMessage_ProducerFailsWithRetryHandlerFailureCallback() throws Exception { AtomicBoolean onSuccess = new AtomicBoolean(false); MockChannel channel = createChannel(new FailFirstMockMessageProducer(5), Arrays.asList(new Service[] {new NullService()})); channel.setMessageErrorHandler(new RetryMessageErrorHandler(4, new TimeInterval(100l, TimeUnit.MILLISECONDS), new NullService())); AdaptrisMessage msg = AdaptrisMessageFactory.getDefaultInstance().newMessage(PAYLOAD_1); StandardWorkflow workflow = (StandardWorkflow) channel.getWorkflowList().get(0); try { start(channel); workflow.onAdaptrisMessage(msg, (m) -> { onSuccess.set(true); }, (m) -> { onSuccess.set(false); }); assertFalse(onSuccess.get()); } finally { stop(channel); } }