Web3.py: eth_subscribe and eth_unsubscribe support

[tsudmi]

Hi,

Are there any plans on implementing eth_subscribe/eth_unsubscribe RPC calls similarly how it's done in web3js?

eth_getFilterChanges consumes a lot of traffic as you need to always poll the node for the changes. With eth_subscribe node will push the updates upon arrival.

[kclowes]

We do not have any plans yet, but will leave this ticket open to indicate that there is interest!

[jerryji]

As of Nov-2019, Infura keeps promoting eth_subscribe to an extreme extent that they are not trying to fix any other websocket issues for web3.py

Here's a week-long agonizing talk I had with them --

https://community.infura.io/t/ropsten-websocket-event-timeout-with-web3-py/1078

Given Infura's direction, could the discussion to support eth_subscribe/eth_unsubscribe be revived?

Thanks.

[om26er]

We are trying to keep our multiple off-chain backends in sync with the blockchain but are having a hard time doing that. We currently have the polling interval set to 300 seconds, maybe we could adjust that a bit. However there are still cases where we could only "afford" a few seconds lag between on-chain and off-chain states. Being able to subscribe to blockchain events will likely make things simpler for us. It will also enable us to simplify our blockchain monitoring code largely.

[tgdn]

Hi, thank you for developing this project.

This issue was opened two years ago, has anything been done to implement subscriptions?

Thanks.

[kclowes]

@tgdn Nope, not yet, but it's fairly high priority at this point. If anyone has the time to put up a PR, I'd be happy to review/help out. That's probably the fastest way to get something in!

[tgdn]

Do you have any idea how you would like this to be built?
I suppose it would be available through the websocket interface.

[kclowes]

I'd like to get some feedback from @pipermerriam because I'm not super familiar with python pub/sub paradigms. One naive implementation might be to add the subscribe and unsubscribe methods to the Eth Module, and then we'd have a similar API to the event examples here. But I'm curious if there's a more elegant way.

/cc @marcgarreau

[marcgarreau]

Also unsure of the most pythonic approach. Web3.js uses the event emitter pattern, e.g.,

myContract.events.MyEvent({})
    .on('data', async function(event) {
        // handle event
    })
    .on('error', console.error);

Looks like a python port exists here. Maybe a different flavor of observer pattern is preferred, though. I yield the remainder of my time to Piper or Carver.

[zipzap888]

im curious if anyone has figured out a way to get eth_subscribe to work with web3py? I just ported a project from web3js because I wanted to work with scikit, only to find that I can't subscribe. I'd love to see this feature implemented, or if someone has an idea of a workaround, that would be great too!

[zipzap888]

here are two examples of pub sub. python-binance-chain has a subscribe feature and an auto reconnect for their websockets connection.
https://github.com/eclipse/paho.mqtt.python#subscribe-unsubscribe
https://github.com/sammchardy/python-binance-chain

[quazzuk]

can't something similar to this be used?

[Fillicia]

I'm not usually one to give tricks but here's a small snippet you can use to subscribe with autobahn/twisted under python

from twisted.internet.protocol import ReconnectingClientFactory
from autobahn.twisted.websocket import WebSocketClientProtocol, \
    WebSocketClientFactory
import json

#This part dictate how the different events are managed with your websocket
class MyClientProtocol(WebSocketClientProtocol):

    def onConnect(self, response):
        print("Server connected: {0}".format(response.peer))

    def onConnecting(self, transport_details):
        print("Connecting; transport details: {}".format(transport_details))
        return None  # ask for defaults

    def onOpen(self):
        print("WebSocket connection open.")
        # Change this part to the subscription you want to get
        self.sendMessage(json.dumps({"jsonrpc":"2.0","id": 1, "method": "eth_subscribe", "params": ["newHeads"]}).encode('utf8'))

    def onMessage(self, payload, isBinary):
        if isBinary:
            print("Binary message received: {0} bytes".format(len(payload)))
        else:
            print("Text message received: {0}".format(payload.decode('utf8')))



    def onClose(self, wasClean, code, reason):
        print("WebSocket connection closed: {0}".format(reason))

#This is to automatically reconnect on failure
class MyClientFactory(WebSocketClientFactory, ReconnectingClientFactory):

    protocol = MyClientProtocol

    def clientConnectionFailed(self, connector, reason):
        print("Client connection failed .. retrying ..")
        self.retry(connector)

    def clientConnectionLost(self, connector, reason):
        print("Client connection lost .. retrying ..")
        self.retry(connector)


if __name__ == '__main__':

    import sys

    from twisted.python import log
    from twisted.internet import reactor

    log.startLogging(sys.stdout)

    factory = MyClientFactory("wss://eth-ropsten.ws.alchemyapi.io/v2/<your API>")

    reactor.connectTCP("eth-ropsten.ws.alchemyapi.io", 80, factory)
    reactor.run()

You would have to manage the response under the onMessage, probably with a json.loads(payload) thingy. I'll let you all discover the rest. Also this is twisted which is similar but different to asyncio so you would have to manage it as such but it's a good start

[ls-lucency]

any updates on the implementation of subscriptions?

[kclowes]

I looked into doing this a while back, and there wasn't really a straightforward way to do it within our existing architecture. It's not on our immediate roadmap but it will probably move up in priority once async gets off the ground since there is so much interest! If anyone wants to take a crack at it, I'd be happy to review/keep things moving!