This is broken, as if the data goes unconsumed, the socket will shut down. The socket buffer management was done by DataEmitterReader above.