Improve shadowing, support async sockets/callbacks in ZMQStream

[minrk]

• can now create alternate socket APIs for existing sockets by passing one socket to another Socket constructor, with e.g. zmq.asyncio.Socket(socket). Also works with shadow: zmq.Socket.shadow(other_socket) instead of needing shadow(socket.underlying). This also preserves a reference to the shadowed object, avoiding close on garbage collection, which could happen if you tried to do e.g. socket = zmq.asyncio.Socket.shadow(ctx.socket(zmq.PUSH).underlying)
• add socket_class argument to ctx.socket, e.g. ctx.socket(zmq.PUSH, socket_class=zmq.asyncio.Socket) for easier creation of explicit socket types in a multi-socket-type situation
• Support async functions in ZMQStream callbacks
• Support async sockets in ZMQStream (by explicitly shadowing with a sync Socket, no matter what kind of Socket it is)

The two seemingly unrelated changes are together because the async socket->sync socket shadow in zmqstream relies on the reference held when shadowing a socket object.

cc @blink1073

In particular, this has consequences for anything that was relying on the fact that on_recv callbacks were passed a Future. This fixes that in that it will get the right argument now, but it means anything that had adopted and assumed the current (wrong) behavior will need to be updated. I'm not sure what the right way to 'fix' this is without breaking anything that assumed the broken behavior.

[blink1073]

Maybe explicitly override if the socket subclasses the asyncio socket, and warn otherwise?

[minrk]

WDYT about warning in the async case? This is where behavior is changing, so a warning makes sense. But also now that shadowing is added, it works fine if the new behavior is what you expect, so a warning might be annoying!

[blink1073]

Maybe instead error out if the original socket is async but the callback is not? That way you're exposing the previously silent bad behavior.

[minrk]

Hm, that's a little tricky since on_recv is called later, and by then the socket has been cast to sync already.

You also can't tell for sure whether a callback is truly async without calling it. (it could be a sync function e.g. via a decorator that schedules an async handler, which is fine, or calls future.add_done_callback).

Maybe the best is to always warn/cast, and make explicit: ZMQStream only accepts zmq.Socket. If you give it anything else, it'll cast to sync, but will ask you to do the casting for next time.

[blink1073]

Fair enough, that works

[minrk]

Went with explicit warning for async sockets, so if it casts, it warns.

Decided to ignore random custom socket classes, since they are rare, and likely implement custom serialization rather than something async. If they are subclasses of the async sockets, it'll the warning path.

Not sure 100%

[blink1073]

LGTM, thanks!