SequenceGeneratorSource extends AbstractSource implements
    PollableSource { @Override public Status process() throws EventDeliveryException { try { getChannelProcessor().processEvent( EventBuilder.withBody(String.valueOf(sequence++).getBytes())); counterGroup.incrementAndGet("events.successful"); } catch (ChannelException ex) { counterGroup.incrementAndGet("events.failed"); } return Status.READY; } SequenceGeneratorSource(); @Override Status process(); @Override void start(); @Override void stop();  }
@Test public void testProcess() throws InterruptedException, LifecycleException, EventDeliveryException { Channel channel = new PseudoTxnMemoryChannel(); Context context = new Context(); context.put("logicalNode.name", "test"); Configurables.configure(source, context); Configurables.configure(channel, context); List<Channel> channels = new ArrayList<Channel>(); channels.add(channel); ChannelSelector rcs = new ReplicatingChannelSelector(); rcs.setChannels(channels); source.setChannelProcessor(new ChannelProcessor(rcs)); for (long i = 0; i < 100; i++) { source.process(); Event event = channel.take(); Assert.assertArrayEquals(String.valueOf(i).getBytes(), new String(event.getBody()).getBytes()); } }