replace `ref.invalid` with proper notifiers that listen to SDK events

[gnunicorn]

And make sure they use the proper pattern of subscribeStream (that is rust internal) and listner.listen for the actual stream.

_listener = XXX.subscribeStream(); // keep it resident in memory
_poller = _listener.listen(