@Override
    @SuppressWarnings("unchecked")
    protected void onInit() throws Exception {
        super.onInit();
        //Stream with a RingBufferProcessor
        this.ringBufferProcessor = RingBufferProcessor.share("xd-reactor", getRingBufferSize());

        //user defined stream processing
        Publisher<?> outputStream = processor.process(Streams.wrap(ringBufferProcessor).env(getEnvironment()));

        outputStream.subscribe(new ChannelForwardingSubscriber());
    }
Construct a new BroadcasterMessageHandler given the reactor based Processor to delegate
processing to.

@param processor The stream based reactor processor
spring-xd-reactor/src/main/java/org/springframework/xd/reactor/BroadcasterMessageHandler.java