channelContext.put("capacity", "1000000");
channelContext.put("keep-alive", "0"); // for faster tests
Channel channel = new MemoryChannel();
Configurables.configure(channel, new Context(channelContext));
Sink sink = new LoggerSink();
sink.setChannel(channel);
sink.start();
DefaultSinkProcessor proc = new DefaultSinkProcessor();
proc.setSinks(Collections.singletonList(sink));
SinkRunner sinkRunner = new SinkRunner(proc);
sinkRunner.start();
ChannelSelector rcs = new ReplicatingChannelSelector();
rcs.setChannels(Collections.singletonList(channel));
ChannelProcessor chp = new ChannelProcessor(rcs);
source.setChannelProcessor(chp);
source.start();
Thread.sleep(5000);
source.stop();
sinkRunner.stop();
sink.stop();
}