@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); } }
@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 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 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); }
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(); }
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; }