Stop consumer from reading on connections closed by producer

[cw9]

When there is a very slow consumer, the producer could fill up the tcp write buffer and time out on tcp writes, which will cause the producer to close the connection and recreate a new one. But there is no way for consumer to know that the existing connection is closed, which means it will continue to read msg from the connection and work on the msg, although all of the work will be wasted as they won't be acked back to the producer, so the messages will be tried again. It would be good to somehow detect the close of the connection and stop processing messages from it.