protected void invokeProcessor(Message<?> message, RingBufferProcessor<Object> reactiveProcessorToUse) {
        // pass the message directly if the input type accepts it, unless the input type is Object
        // this restricts the branch to Message and its subinterfaces/implementations
        if (!Object.class.equals(inputType) && ClassUtils.isAssignable(inputType, message.getClass())) {
            reactiveProcessorToUse.onNext(message);
        } else if (ClassUtils.isAssignable(inputType, message.getPayload().getClass())) {
            reactiveProcessorToUse.onNext(message.getPayload());
        } else {
            throw new MessageHandlingException(message, "Processor signature does not match [" + message.getClass()
                + "] or [" + message.getPayload().getClass() + "]");
        }
    }
Returns the input type of the stream that is to be processed
spring-xd-reactor/src/main/java/org/springframework/xd/reactor/AbstractReactorMessageHandler.java