message.* triggers _every_ message handler

This makes every type-based handler (message.something, for a "type:
something" in the JSON) fire for every message pushed over the bus!
Ensure orderly cleanup if the socket is closed
on_error is not an accessor, and would return a sub{} if it were