PooledSplitJoinService extends ServiceImp implements EventHandlerAware, ServiceWrapper { @Override public void registerEventHandler(EventHandler eh) { eventHandler = eh; }  @Override void prepare(); @Override void start(); @Override void stop(); @Override void doService(AdaptrisMessage msg); @Override void registerEventHandler(EventHandler eh); @Override Service[] wrappedServices();  }
@Test public void testService_Events() throws Exception { PooledSplitJoinService service = new PooledSplitJoinService(); MockMessageProducer events = new MockMessageProducer(); service.setAggregator(new AppendingMessageAggregator()); service.setSplitter(new LineCountSplitter(1)); service.setPoolsize(10); service.setTimeout(new TimeInterval(10L, TimeUnit.SECONDS)); service.setService(new WaitService(new TimeInterval(100L, TimeUnit.MILLISECONDS), true)); service.setSendEvents(true); service.registerEventHandler(LifecycleHelper.initAndStart(new DefaultEventHandler(events))); AdaptrisMessage msg = createLineCountMessageInput(50); ExampleServiceCase.execute(service, msg); List<String> result = IOUtils.readLines(msg.getReader()); assertEquals(50 * 2, result.size()); assertEquals(50, events.getMessages().size()); }
@Test(expected = ServiceException.class) public void testService_WithException() throws Exception { PooledSplitJoinService service = new PooledSplitJoinService(); service.setAggregator(new AppendingMessageAggregator()); service.setSplitter(new LineCountSplitter(1)); service.setPoolsize(10); service.setTimeout(new TimeInterval(5L, TimeUnit.SECONDS)); service.setService(new ThrowExceptionService(new ConfiguredException("always-fail"))); service.registerEventHandler(LifecycleHelper.initAndStart(new DefaultEventHandler())); AdaptrisMessage msg = createLineCountMessageInput(50); try { ExampleServiceCase.execute(service, msg); } catch (ServiceException e) { assertEquals("always-fail", e.getMessage()); throw e; } }
@Test public void testService_DidWorkSuccessfully() throws Exception { PooledSplitJoinService service = new PooledSplitJoinService(); service.setServiceErrorHandler( new NoExceptionIfWorkDone().withMetadataKey(NoExceptionIfWorkDone.DEFAULT_METADATA_KEY)); service.setSplitter(new LineCountSplitter(1)); service.setService( asCollection(new WaitService(new TimeInterval(10L, TimeUnit.MILLISECONDS), false), new MockExceptionStrategyService(MockExceptionStrategyService.MODE.NEUTRAL))); service.setPoolsize(10); service.setTimeout(new TimeInterval(5L, TimeUnit.SECONDS)); service.setAggregator(new AppendingMessageAggregator()); service.registerEventHandler(LifecycleHelper.initAndStart(new DefaultEventHandler())); AdaptrisMessage msg = createLineCountMessageInput(50); execute(service, msg); List<String> result = IOUtils.readLines(msg.getReader()); assertEquals(50 * 2, result.size()); }
@Test(expected = ServiceException.class) public void testService_DidNoWork() throws Exception { PooledSplitJoinService service = new PooledSplitJoinService(); service.setServiceErrorHandler(new NoExceptionIfWorkDone().withMetadataKey(NoExceptionIfWorkDone.DEFAULT_METADATA_KEY)); service.setSplitter(new LineCountSplitter(1)); service.setService(asCollection(new MockExceptionStrategyService(MockExceptionStrategyService.MODE.ERROR))); service.setPoolsize(10); service.setTimeout(new TimeInterval(5L, TimeUnit.SECONDS)); service.setAggregator(new AppendingMessageAggregator()); service.registerEventHandler(LifecycleHelper.initAndStart(new DefaultEventHandler())); AdaptrisMessage msg = createLineCountMessageInput(50); execute(service, msg); }
@Test public void testService_DidSomeWork() throws Exception { PooledSplitJoinService service = new PooledSplitJoinService(); service.setServiceErrorHandler( new NoExceptionIfWorkDone().withMetadataKey(NoExceptionIfWorkDone.DEFAULT_METADATA_KEY)); service.setSplitter(new LineCountSplitter(1)); service.setService( asCollection(new MockExceptionStrategyService(MockExceptionStrategyService.MODE.MIXED))); service.setPoolsize(10); service.setTimeout(new TimeInterval(5L, TimeUnit.SECONDS)); service.setAggregator(new AppendingMessageAggregator()); service.registerEventHandler(LifecycleHelper.initAndStart(new DefaultEventHandler())); AdaptrisMessage msg = createLineCountMessageInput(50); execute(service, msg); List<String> result = IOUtils.readLines(msg.getReader()); assertEquals(50 * 2, result.size()); }
@Test public void testService_IgnoreExceptions() throws Exception { PooledSplitJoinService service = new PooledSplitJoinService(); service.setServiceErrorHandler(new IgnoreAllExceptions()); service.setSplitter(new LineCountSplitter(1)); service.setService( asCollection(new MockExceptionStrategyService(MockExceptionStrategyService.MODE.ERROR))); service.setPoolsize(10); service.setTimeout(new TimeInterval(5L, TimeUnit.SECONDS)); service.setAggregator(new AppendingMessageAggregator()); service.registerEventHandler(LifecycleHelper.initAndStart(new DefaultEventHandler())); AdaptrisMessage msg = createLineCountMessageInput(50); execute(service, msg); List<String> result = IOUtils.readLines(msg.getReader()); assertEquals(50 * 2, result.size()); }