@Test public void testWrongMessage() throws Exception { AdaptrisMessage message = AdaptrisMessageFactory.getDefaultInstance().newMessage(PAYLOAD_1); forEach.doService(message); verify(mock, times(1)).doService(message); }
@Override public void doService(AdaptrisMessage msg) throws ServiceException { ThreadPoolExecutor executor = null; try { log.info("Starting for-each"); if (!(msg instanceof MultiPayloadAdaptrisMessage)) { log.warn("Message [{}] is not a multi-payload message!", msg.getUniqueId()); iterate(msg); } else { MultiPayloadAdaptrisMessage message = (MultiPayloadAdaptrisMessage)msg; int threads = threadCount; if (threads == 0) { threads = message.getPayloadCount(); } log.trace("Using {} thread{}", threads, threads > 1 ? "s" : ""); executor = (ThreadPoolExecutor)Executors.newFixedThreadPool(threads); for (String id : message.getPayloadIDs()) { try { message.switchPayload(id); AdaptrisMessage each = DefaultMessageFactory.getDefaultInstance().newMessage(message, null); each.setPayload(message.getPayload()); executor.execute(() -> iterate(each)); } catch (CloneNotSupportedException e) { log.error("Could not clone message [{}]", id, e); } } } } finally { if (executor != null) { executor.shutdown(); try { while (!executor.awaitTermination(1, TimeUnit.MILLISECONDS)) { } } catch (InterruptedException e) { log.warn("Interrupted while waiting for tasks to finish!", e); } } log.info("Finished for-each"); } }
ForEach extends ServiceImp { @Override public void doService(AdaptrisMessage msg) throws ServiceException { ThreadPoolExecutor executor = null; try { log.info("Starting for-each"); if (!(msg instanceof MultiPayloadAdaptrisMessage)) { log.warn("Message [{}] is not a multi-payload message!", msg.getUniqueId()); iterate(msg); } else { MultiPayloadAdaptrisMessage message = (MultiPayloadAdaptrisMessage)msg; int threads = threadCount; if (threads == 0) { threads = message.getPayloadCount(); } log.trace("Using {} thread{}", threads, threads > 1 ? "s" : ""); executor = (ThreadPoolExecutor)Executors.newFixedThreadPool(threads); for (String id : message.getPayloadIDs()) { try { message.switchPayload(id); AdaptrisMessage each = DefaultMessageFactory.getDefaultInstance().newMessage(message, null); each.setPayload(message.getPayload()); executor.execute(() -> iterate(each)); } catch (CloneNotSupportedException e) { log.error("Could not clone message [{}]", id, e); } } } } finally { if (executor != null) { executor.shutdown(); try { while (!executor.awaitTermination(1, TimeUnit.MILLISECONDS)) { } } catch (InterruptedException e) { log.warn("Interrupted while waiting for tasks to finish!", e); } } log.info("Finished for-each"); } } }
ForEach extends ServiceImp { @Override public void doService(AdaptrisMessage msg) throws ServiceException { ThreadPoolExecutor executor = null; try { log.info("Starting for-each"); if (!(msg instanceof MultiPayloadAdaptrisMessage)) { log.warn("Message [{}] is not a multi-payload message!", msg.getUniqueId()); iterate(msg); } else { MultiPayloadAdaptrisMessage message = (MultiPayloadAdaptrisMessage)msg; int threads = threadCount; if (threads == 0) { threads = message.getPayloadCount(); } log.trace("Using {} thread{}", threads, threads > 1 ? "s" : ""); executor = (ThreadPoolExecutor)Executors.newFixedThreadPool(threads); for (String id : message.getPayloadIDs()) { try { message.switchPayload(id); AdaptrisMessage each = DefaultMessageFactory.getDefaultInstance().newMessage(message, null); each.setPayload(message.getPayload()); executor.execute(() -> iterate(each)); } catch (CloneNotSupportedException e) { log.error("Could not clone message [{}]", id, e); } } } } finally { if (executor != null) { executor.shutdown(); try { while (!executor.awaitTermination(1, TimeUnit.MILLISECONDS)) { } } catch (InterruptedException e) { log.warn("Interrupted while waiting for tasks to finish!", e); } } log.info("Finished for-each"); } }  }
ForEach extends ServiceImp { @Override public void doService(AdaptrisMessage msg) throws ServiceException { ThreadPoolExecutor executor = null; try { log.info("Starting for-each"); if (!(msg instanceof MultiPayloadAdaptrisMessage)) { log.warn("Message [{}] is not a multi-payload message!", msg.getUniqueId()); iterate(msg); } else { MultiPayloadAdaptrisMessage message = (MultiPayloadAdaptrisMessage)msg; int threads = threadCount; if (threads == 0) { threads = message.getPayloadCount(); } log.trace("Using {} thread{}", threads, threads > 1 ? "s" : ""); executor = (ThreadPoolExecutor)Executors.newFixedThreadPool(threads); for (String id : message.getPayloadIDs()) { try { message.switchPayload(id); AdaptrisMessage each = DefaultMessageFactory.getDefaultInstance().newMessage(message, null); each.setPayload(message.getPayload()); executor.execute(() -> iterate(each)); } catch (CloneNotSupportedException e) { log.error("Could not clone message [{}]", id, e); } } } } finally { if (executor != null) { executor.shutdown(); try { while (!executor.awaitTermination(1, TimeUnit.MILLISECONDS)) { } } catch (InterruptedException e) { log.warn("Interrupted while waiting for tasks to finish!", e); } } log.info("Finished for-each"); } }  ThenService getThen(); void setThen(ThenService thenService); Integer getThreadCount(); void setThreadCount(Integer threadCount); @Override void doService(AdaptrisMessage msg); @Override void prepare(); @Override void start(); @Override void stop(); }
ForEach extends ServiceImp { @Override public void doService(AdaptrisMessage msg) throws ServiceException { ThreadPoolExecutor executor = null; try { log.info("Starting for-each"); if (!(msg instanceof MultiPayloadAdaptrisMessage)) { log.warn("Message [{}] is not a multi-payload message!", msg.getUniqueId()); iterate(msg); } else { MultiPayloadAdaptrisMessage message = (MultiPayloadAdaptrisMessage)msg; int threads = threadCount; if (threads == 0) { threads = message.getPayloadCount(); } log.trace("Using {} thread{}", threads, threads > 1 ? "s" : ""); executor = (ThreadPoolExecutor)Executors.newFixedThreadPool(threads); for (String id : message.getPayloadIDs()) { try { message.switchPayload(id); AdaptrisMessage each = DefaultMessageFactory.getDefaultInstance().newMessage(message, null); each.setPayload(message.getPayload()); executor.execute(() -> iterate(each)); } catch (CloneNotSupportedException e) { log.error("Could not clone message [{}]", id, e); } } } } finally { if (executor != null) { executor.shutdown(); try { while (!executor.awaitTermination(1, TimeUnit.MILLISECONDS)) { } } catch (InterruptedException e) { log.warn("Interrupted while waiting for tasks to finish!", e); } } log.info("Finished for-each"); } }  ThenService getThen(); void setThen(ThenService thenService); Integer getThreadCount(); void setThreadCount(Integer threadCount); @Override void doService(AdaptrisMessage msg); @Override void prepare(); @Override void start(); @Override void stop();  }