- Producer: Add notion of "connected" and check before allowing to Send.

- Protocol: Break from CorrelateResponseLoop if connection is closeed.
- Consumer - specify to ObserveOn default scheduler instead of kafka-scheduler.

- Tests - update tests to properly check that the thread is not the kafka-scheduler thread.