A new type MessageHeaderAccesssor provides read/write access to
MessageHeaders along with typed getter/setter methods along the lines
of the existing MessageBuilder methods (internally MessageBuilder
merely delegates to MessageHeaderAccessor). This class is extensible
with sub-classes expected to provide typed getter/setter methods for
specific categories of message headers.

NativeMessageHeaderAccessor is one specific sub-class that further
provides read/write access to headers from some external message
source (e.g. STOMP headers). Native headers are stored in a separate
MultiValueMap and kept under a specific key.

Similar to @ExceptionHandler but for message processing. Such a method
can send messages to both the message broker channel and the client
channel provided the client is subscribed to the target destination.

The "system" STOMP session is established at startup and can be used
to send messages without a client session, e.g. to support broadcasting
from a REST/HTTP handler method.

MessageHolder holds the currently processed message in a ThreadLocal,
which allows PubSubMessageBuilder to automatically add a session id
to messages to be sent.

Rename to PubSubHeaderAccessor and StompHeaderAccessor
Move the renamed classes to support packages

Remove fromPayloadAndHeaders from MessageBuilder, just use
withPayload(..).copyHeaders(..) instead.

The use of an AtomicBoolean and no lock meant that it was possible
for a message to be queued and then never be flushed and sent to the
broker:

1. On t1, a message is received and isConnected is false. The message
   will be queued.
2. On t2, CONNECTED is received from the broker. isConnected is set
   to true, the queue is drained and the queued messages are forwarded
3. On t1, the message is added to the queue

To fix this, checking that isConnected is false (step 1 above) and the
queueing of a message (step 3 above) need to be performed as a unit
so that the flushing of the queued messages can't be interleaved. This
is achieved by synchronizing on a monitor and performing steps 1
and 3 and synchronizing on the same monitor while performing step 2.

The monitor is held while the messages are actually being forwarded
to the broker. An alternative would be to drain the queue into
a local variable, release the monitor, and then forward the messages.
The main advantage of this alternative is that the monitor is held for
less time. It also reduces the theoretical risk of deadlock by not
holding the monitor while making an alien call. The downside of the
alternative is that it may lead to messages being forwarded out of
order. For this reason the alternative approach was rejected.

When an annotated handler returns a Message from a @SubscribeEvent
or @MessageMapping method and it contains no destination in its
headers, use the received message's destination as the response
message's destination.

Without generics, extending AbstractPubSubChannelRegistry and using
a custom Message type requires some unpleasant casting and suppression
of warnings. By genericizing PubSubChannelRegistry and
AbstractPubSubChannelRegistry these problems can be avoided.

To improve compatibility between Spring's messaging classes and
Spring Integration, the type of Message that is created has been made
pluggable through the introduction of a factory abstraction;
MessageFactory.

By default a MessageFactory is provided that will create
org.springframework.messaging.GenericMessage instances, however this
can be replaced with an alternative implementation. For example,
Spring Integration can provide an implementation that creates
org.springframework.integration.message.GenericMessage instances.

This control over the type of Message that's created allows messages
to flow from Spring messaging code into Spring Integration code without
any need for conversion. In further support of this goal,
MessageChannel, MessageHandler, and SubscribableChannel have been
genericized to make the Message type that they deal with more
flexible.

Correct the Assert imports

Import org.springframework.util.Assert rather than reactor.util.Assert