@SuppressWarnings("unchecked")
    private RingBufferProcessor<Object> getReactiveProcessor(Message<?> message) {
        final Object idToUse = partitionExpression.getValue(evaluationContext, message, Object.class);
        if (logger.isDebugEnabled()) {
            logger.debug("Partition Expression evaluated to " + idToUse);
        }
        RingBufferProcessor<Object> reactiveProcessor = reactiveProcessorMap.get(idToUse);
        if (reactiveProcessor == null) {
            RingBufferProcessor<Object> existingReactiveProcessor =
                    reactiveProcessorMap.putIfAbsent(idToUse,
                            RingBufferProcessor.share("xd-reactor-partition-" + idToUse, getRingBufferSize()));
            if (existingReactiveProcessor == null) {
                reactiveProcessor = reactiveProcessorMap.get(idToUse);
                //user defined stream processing
                Publisher<?> outputStream = processor.process(Streams.wrap(reactiveProcessor).env(getEnvironment()));

                outputStream.subscribe(new ChannelForwardingSubscriber());

            }
            else {
                reactiveProcessor = existingReactiveProcessor;
            }
        }
        return reactiveProcessor;
    }
Construct a new MessageHandler given the reactor based Processor to delegate
processing to and a partition expression.

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