Websockets support in web3j

[conor10]

web3j doesn't currently support web socket connections to Geth and Parity. It would be useful to include this functionality.

• Parity web sockets
• Geth PUB SUB

[mushketyk]

If nobody is working on this issue, I would like to implement this.

[iikirilov]

Infura ws info:

INFURA/infura#73

[mushketyk]

Sorry for a long delay.

Since this issue is tagged with mini-project I would like to share my design proposal before I attempt to work on it.

As a result of this issue we should be able to subscribe to updates from the Ethereum network using the WebSocket protocol (example from the go-ethereum doc):

// create subscription
>> {"id": 1, "method": "eth_subscribe", "params": ["newHeads", {}]}
<< {"jsonrpc":"2.0","id":1,"result":"0xcd0c3e8af590364c09d0fa6a1210faf5"}

// incoming notifications
<< {"jsonrpc":"2.0","method":"eth_subscription","params":{"subscription":"0xcd0c3e8af590364c09d0fa6a1210faf5","result":{"difficulty":"0xd9263f42a87",<...>, "uncles":[]}}}
<< {"jsonrpc":"2.0","method":"eth_subscription","params":{"subscription":"0xcd0c3e8af590364c09d0fa6a1210faf5","result":{"difficulty":"0xd90b1a7ad02", <...>, "uncles":["0x80aacd1ea4c9da32efd8c2cc9ab38f8f70578fcd46a1a4ed73f82f3e0957f936"]}}}

Essentially we need to call the eth_subscribe and if subscription was successful JSON RPC would return a subscription id and send will send notifications for this subscription id.

Subscriptions can be canceled by calling the eth_unsubscribe method and providing the subscption id:

// cancel subscription
>> {"id": 1, "method": "eth_unsubscribe", "params": ["0xcd0c3e8af590364c09d0fa6a1210faf5"]}
<< {"jsonrpc":"2.0","id":1,"result":true}

This API allows to subscribe to following updates:

• newHeads - an update is sent when a new head is generated. Does not accept any additional parameters.
• logs - returns logs that are added to new blocks. Has two optional parameters: address and topics.
• newPendingTransactions - an update is sent for every new transaction is added to the pending state. Does not accept any additional parameters.
• syncing - an update is sent when a node starts/stops syncing. Does not accept any additional parameters.

Since WebSockets interface is just an alternative transport for JSON RPC, I propose to add another implementation of the Web3jService interface, called WebSocketsService that will use WebSockets as transport. To allow clients of the Web3jService interface to subscribe to a stream of events we need to add two more methods to the Web3jService interface:

public interface Web3jService {
    <T extends Response> T send(
            Request request, Class<T> responseType) throws IOException;

    <T extends Response> CompletableFuture<T> sendAsync(
            Request request, Class<T> responseType);

    /**
     * Subscribe to a sream of events
     */
    default Observable<Event<T>> subscribe(SubsribtionRequest request, Class<T> eventType) {
        throw new UnsupportedOperationException();
    }

    /**
     * Returns true if an implementation supports subscription, returns false otherwise
     */
    default boolean supportsSubscription() {
        return false;
    }
}

We need two separate methods because subsciptions are not supported by HTTP transport and we need to distinguish what methods are supported by a particular implementation:

$ curl --data '{"method":"eth_subscribe","params":["newHeads"],"id":1,"jsonrpc":"2.0"}' -H "Content-Type: application/json" -X POST localhost:8545
{"jsonrpc":"2.0","id":1,"error":{"code":-32000,"message":"notifications not supported"}}

By default implementations of the Web3jService would throw an instance of the UnsupportedOperationException if the subscribe method is called, but the WebSocketsService will use the WebSocket library to implement it.

JsonRpc2_0Web3j already provides methods similar to methods provided by eth_subscribe but it does not have support for all updates. To allow clients to subscribe to WebSocket events I suggest to add following methods to JsonRpc2_0Web3j and JsonRpc2_0Rx that would match updates provided by the WebSockets API:

• subscirbeToNewHeads
• subscribeToLogs
• subscribeToNewPendingTransactions
• subscribeToSyncingStatus

Currently JsonRpc2_0Rx relies on polling to implement its methods. To implement new methods, I suggest to turn JsonRpc2_0Rx into an abstract class and provide two implementations:

• JsonRpc2_0RxPolling - implements its methods without using subscriptions
• JsonRpc2_0RxSubscription - relies on the subscription feature

To create a proper instance of the JsonRpc2_0Rx in the JsonRpc2_0Web3j constructor, we would have to create a factory that would decide what instance to create depending on the capabilities of the Web3jService implementation. So this line:

this.web3jRx = new JsonRpc2_0Rx(this, scheduledExecutorService);

will be replaced with something like:

this.web3jRx = JsonRpc2_0RxFactory.instance()
                .withWeb3j(this)
                .withPollingInterval(pollingInterval)
                .withExecutorService(scheduledExecutorService)
                .build()

What do you think about this? Does it make sense? Did you have other design ideas in mind?

[conor10]

@mushketyk I really like your suggestions here, it's well thought out. The fluent interface is a nice idea too.

The only additional considerations I have are:

1. Do you have a client side library in mind for websocket support? I'm keen to avoid anything heavyweight such as Spring.
2. It will be great if the NEW WebSocketsService just works with the existing filters generated in the smart contract wrappers, so its pluggable which type is used (appreciate you may already have thought this through).

[mushketyk]

Do you have a client side library in mind for websocket support? I'm keen to avoid anything heavyweight such as Spring.

What do you think about this one: https://github.com/TooTallNate/Java-WebSocket ? It has no dependencies, and it seems to be quite popular (judging by the number of GitHub stars :) )

It will be great if the NEW WebSocketsService just works with the existing filters generated in the smart contract wrappers, so its pluggable which type is used

Do you mean code like this (generated code from SolidityFunctionWrapperTest):

public rx.Observable<TransferEventResponse> transferEventObservable(org.web3j.protocol.core.methods.request.EthFilter filter) {
    return web3j.ethLogObservable(filter).map(new rx.functions.Func1<org.web3j.protocol.core.methods.response.Log, TransferEventResponse>() {
        ...
    };
}

It should support existing filters since WebSockets transport supports all RPC methods that are available via HTTP. I don't think we can use WebSockets subscriptions to implement all methods that that Web3jRx provides since the eth_subscribe provides a limited set of data streams, but it should work nonetheless.

[Qkyrie]

How's this implementation going @mushketyk? :)

[mushketyk]

Hey @Qkyrie The implementation is almost finished. I've just updated my PR and waiting for a review: #458

[WarSame]

@mushketyk Thank you for your improvement!

I am interested in helping to get this working on Android. However, there is no documentation describing the process of making an release ready for Android. What makes a release ready for Android?

I will also update the documentation, mentioning that filtering is possible with Infura, once it is possible on Android.

[mushketyk]

@WarSame Sorry, but I am not sure how to make a release ready for Android. You need to ask @conor10 about this.

[WarSame]

@mushketyk Thanks. Also, I thought your changes made it into 3.4.0 according to the notes. They mention "Support for WebSocket to listen Log Events #392", where #392 points right here, #261. However, I went to try the latest release and didn't find WebSocketService within the core protocols. Did I miss something?

@conor10 Is there a published process to make a release Android-ready?

[macalinao]

Could someone publish the new web3j with websocket support? Thanks!

[pradyuman]

What's the timeline for this getting published in a release?

[WarSame]

@conor10 Could you please let me know how I can help this get released? There are a few features I want to develop that are blocked by it.

[dbryan0516]

Noting here that we need to add basic auth support for websockets as aligned with other web3 implementations

[iikirilov]

@mushketyk @dbryan0516 Is there anything in the scope of this issue we would like to add?

@WarSame regarding android release procedure #769

[dbryan0516]

@iikirilov Nothing on my end