@Override
    protected void handleMessageInternal(Message<?> message) throws Exception {
        if (ClassUtils.isAssignable(inputType.getRawClass(), message.getClass())) {
            subject.onNext(message);
        } else if (ClassUtils.isAssignable(inputType.getRawClass(), message.getPayload().getClass())) {
            subject.onNext(message.getPayload());
        } else {
            throw new MessageHandlingException(message, "Processor signature does not match [" + message.getClass()
                    + "] or [" + message.getPayload().getClass() + "]");
        }
    }
Adapts the item at a time delivery of a {@link org.springframework.messaging.MessageHandler}
by delegating processing to a {@link Observable}.
<p/>
The outputStream of the processor is used to create a message and send it to the output channel.  If the
input channel and output channel are connected to the MessageBus, then data delivered to the input stream via
a call to onNext is invoked on the dispatcher thread of the message bus and sending a message to the output
channel will involve IO operations on the message bus.
<p/>
The implementation uses a SerializedSubject.  This has the advantage that the state of the Observabale
can be shared across all the incoming dispatcher threads that are invoking onNext.  It has the disadvantage
that processing and sending to the output channel will execute serially on one of the dispatcher threads.
<p/>
The use of this handler makes for a very natural first experience when processing data.  For example given
the stream <code></code>http | rxjava-processor | log</code> where the <code>rxjava-processor</code> does does a
<code>buffer(5)</code> and then produces a single value.  Sending 10 messages to the http source will
result in 2 messages in the log, no matter how many dispatcher threads are used.
<p/>
You can modify what thread the outputStream subscriber, which does the send to the output channel,
will use by explicitly calling <code>observeOn</code> before returning the outputStream from your processor.
<p/>
Use {@link org.springframework.xd.rxjava.MultipleSubjectMessageHandler} for concurrent execution on dispatcher
threads spread across across multiple Observables.
All error handling is the responsibility of the processor implementation.

@author Mark Pollack
spring-xd-rxjava/src/main/java/org/springframework/xd/rxjava/SubjectMessageHandler.java