PoolingWorkflow extends WorkflowImp { @Override public void onAdaptrisMessage(final AdaptrisMessage msg, Consumer<AdaptrisMessage> success, Consumer<AdaptrisMessage> failure) { ListenerCallbackHelper.prepare(msg, success, failure); if (!obtainChannel().isAvailable()) { handleChannelUnavailable(msg); } else { onMessage(msg); } } PoolingWorkflow();  PoolingWorkflow(String uniqueId); void setPoolSize(Integer i); Integer getPoolSize(); int poolSize(); long threadLifetimeMs(); long shutdownWaitTimeMs(); TimeInterval getThreadKeepAlive(); void setThreadKeepAlive(TimeInterval interval); TimeInterval getShutdownWaitTime(); void setShutdownWaitTime(TimeInterval interval); @Override void onAdaptrisMessage(final AdaptrisMessage msg, Consumer<AdaptrisMessage> success, Consumer<AdaptrisMessage> failure); @Override synchronized void handleBadMessage(AdaptrisMessage msg); @Override synchronized void doProduce(AdaptrisMessage msg); @Override synchronized void handleProduceException(); Integer getThreadPriority(); void setThreadPriority(Integer i); Integer getMinIdle(); void setMinIdle(Integer i); int minIdle(); Integer getMaxIdle(); void setMaxIdle(Integer i); int maxIdle(); int threadPriority(); TimeInterval getInitWaitTime(); void setInitWaitTime(TimeInterval t); long initWaitTimeMs(); int currentObjectPoolCount(); int currentlyActiveObjects(); int currentlyIdleObjects(); int currentThreadPoolCount(); static final int DEFAULT_MAX_POOLSIZE; static final int DEFAULT_MIN_IDLE; static final int DEFAULT_MAX_IDLE; }
@Test public void testHandleChannelUnavailable() throws Exception { final MockChannel channel = createChannel(); PoolingWorkflow wf = (PoolingWorkflow) channel.getWorkflowList().get(0); MockMessageProducer prod = (MockMessageProducer) wf.getProducer(); try { wf.setChannelUnavailableWaitInterval(new TimeInterval(1200L, TimeUnit.MILLISECONDS)); start(channel); AdaptrisMessage msg = AdaptrisMessageFactory.getDefaultInstance().newMessage(""); Timer t = new Timer(true); channel.toggleAvailability(false); t.schedule(new TimerTask() { @Override public void run() { channel.toggleAvailability(true); } }, 500); wf.onAdaptrisMessage(msg); waitForMessages(prod, 1); assertEquals(1, prod.messageCount()); } finally { stop(channel); } }
@Test public void testHandleChannelUnavailableForever() throws Exception { MockMessageProducer producer = new MockMessageProducer(); final MockChannel channel = createChannel(); PoolingWorkflow workflow = (PoolingWorkflow) channel.getWorkflowList().get(0); MockMessageProducer meh = new MockMessageProducer(); try { workflow.setChannelUnavailableWaitInterval(new TimeInterval(100L, TimeUnit.MILLISECONDS)); workflow.setMessageErrorHandler(new StandardProcessingExceptionHandler( new ServiceList(new ArrayList<Service>(Arrays.asList(new Service[] { new StandaloneProducer(meh) }))))); AdaptrisMessage msg = AdaptrisMessageFactory.getDefaultInstance().newMessage(PAYLOAD_1); start(channel); channel.toggleAvailability(false); workflow.onAdaptrisMessage(msg); assertEquals("Make none produced", 0, producer.getMessages().size()); assertEquals(1, meh.getMessages().size()); } finally { stop(channel); } }
@Test public void testOnMessage_SuccessCallback() throws Exception { AtomicBoolean onSuccess = new AtomicBoolean(false); MockChannel channel = createChannel(); AdaptrisMessage msg = AdaptrisMessageFactory.getDefaultInstance().newMessage(PAYLOAD_1); PoolingWorkflow wf = (PoolingWorkflow) channel.getWorkflowList().get(0); MockMessageProducer prod = (MockMessageProducer) wf.getProducer(); try { start(channel); wf.onAdaptrisMessage(msg, (m) -> { onSuccess.set(true); }); waitForMessages(prod, 1); assertTrue(onSuccess.get()); } finally { stop(channel); } }