Service channels & websockets only

[robertmrk]

Service channels & websockets only

[robertmrk]

Hi @boyan-soubachov

Sure, I'd love to help you, and in the meantime we might discover/fix some bugs as well. 😉

1. The library supports the long-polling and websocket transport types. If by websocket-only connections you mean that you would like to force the Client to always chose the websocket transport during a handshake then this is already supported.

client = Client("http://example.com/cometd",
                connection_types=ConnectionType.WEBSOCKET)

2. Service channels or in other words the request/response style of communication is also supported. For sending messages to and receiveng responses from service channels you can use the Client.publish method. If you take a look at the example chat client implementation that communicates with the chat demo service, you can see that it publishes to a service channel called /service/members to join a chat room.

So I think everything you want to use is already supported.
It's a bit difficult to read your logs, since it's not clear which message comes from the library and which one comes from your application, or if the library level log messages has been modified or not.
I would advise to try to use the library without any modifications, and configure some detailed debug level logging, and send me your logs if you encounter any issues with some minimal code examples.

[boyan-soubachov]

Hey @robertmrk ,

1. For sure, but looking at the _negotiate_transport() method in client.py the client first attempts a long-polling connection and then switches to a websocket one. In my case (still not confirmed 100%) the server only runs on websockets so doing an HTTP connection to the server results in a Bayeux error.

I will still dig just to make sure I'm not bypassing an HTTP/long-polling endpoint by mistake but if I'm not, my thought here was to detect if the connection URL's protocol is non-HTTP (e.g. wss://) to skip the long-polling handshake and go straight to the desired transport type.

2. I see. I just thought that because you have SERVICE_CHANNEL_PREFIX defined there's some special communication that needs to happen (e.g. like meta channels) instead of just forcing it over .publish()

Lastly, I've cleared up the logs here. -> signifies client to server and <- server to client:

Connecting to url: wss://tasty.dxfeed.com/live/cometd

Opening client with connection types ['websocket', 'long-polling'] ...

=> [{'channel': <MetaChannel.HANDSHAKE: '/meta/handshake'>, 'version': '1.0', 'supportedConnectionTypes': ['websocket', 'long-polling'], 'minimumVersion': '1.0', 'id': '0', 'ext': {'com.devexperts.auth.AuthToken': '==REDACTED=='}}]

<= [{"minimumVersion":"1.0","clientId":"==REDACTED==","supportedConnectionTypes":["websocket"],"advice":{"interval":0,"timeout":30000,"reconnect":"retry"},"channel":"/meta/handshake","id":"0","version":"1.0","successful":true}]

Connection types supported by the server: ['websocket']

=> [{'channel': <MetaChannel.CONNECT: '/meta/connect'>, 'clientId': '==REDACTED==', 'connectionType': 'websocket', 'id': '0'}]

<= [{"channel":"/meta/connect","id":"0","successful":true}]

Connect task finished with: {'channel': '/meta/connect', 'id': '0', 'successful': True}

Client opened with connection_type 'websocket'

Connection setup completed!

=> [{'channel': <MetaChannel.CONNECT: '/meta/connect'>, 'clientId': '==REDACTED==', 'connectionType': 'websocket', 'id': '1'}]

Adding subscription: {'Quote': ['/ES']}

[dxFeed] sending: {'reset': True, 'add': {'Quote': ['/ES']}}

=> [{'channel': '/service/sub', 'clientId': '==REDACTED==', 'data': {'reset': True, 'add': {'Quote': ['/ES']}}, 'id': '2'}]

<= [{"channel":"/service/sub","id":"2","error":"402::Unknown client","successful":false}]

The ==REDACTED== parts are just secret text-values that I have confirmed work.

Thanks for the help!

P.S. The server I'm connecting to runs Jetty

[robertmrk]

As far as I know (somebody please correct me if I'm wrong here), there is no such thing as a bayeux server without support for long-polling.
Handshake request description in the specs.:

A Bayeux client initiates a connection negotiation by sending a message to the /meta/handshake channel.

In case of HTTP same domain connections, the handshake requests MUST be sent to the server using the long-polling transport, while for cross domain connections the handshake request MAY be sent with the long-polling transport and failing that with the callback-polling transport.

What you're actually trying to achieve here is to skip the connection negotiation step entirely, which I think is not feasible.
I think you should find out what's the HTTP endpoint of your server and use that in the client, while forcing it to chose the websocket transport during the connection negotiation step, by setting connection_types=ConnectionType.WEBSOCKET.

[robertmrk]

BTW. There is nothing special about service channels, at least not on the client side.
I'm not forcing anything here, all other client implementations use the publish method to send these messages.
They only have a specialized handling on the server side. Namely, that the response for messages received on service channels is returned in the same request/connection and that these messages are not broadcasted to other clients.

[boyan-soubachov]

Got it, thank you. I'll do some more digging on my side. Hopefully figuring out the first point will automatically solve the second.

[boyan-soubachov]

Hey @robertmrk , so an update:

I went back to basics and did some more investigation; I opened a basic test/debug site from the same service provider I'm trying to connect to and tried to do a 'by-the-example' aiocometd connection to them.

1. If you open https://tools.dxfeed.com/webservice/debug-console.jsp and inspect the page console, you can see that when you create a profile subscription for a symbol (e.g. '/ES'), you'll see their javascript cometd client trying to connect to https://tools.dxfeed.com/webservice/cometd, e.g.:
   

This connects successfully and looking at the Chromium traffic inspector you can see the websocket connection formed and working.

2. So then I created a lightweight test file to try and use aiocometd to connect to that same endpoint and just listen to any messages (i.e. just see that the connection comes up). I used the following code: https://gist.github.com/boyan-soubachov/65c3531ab7424ec7645bf250b696abd5 to try and do that.
   I then got the same error (if I add some more logging to the _negotiate_transport() method in base.py) as previously with the server telling me "Unknown Bayeux Transport" when it tries to make the long poll connection first.

I would really appreciate it if you could have a look and give me some insight as to why that may be the case and what the best steps forward are?

Thank you!

[robertmrk]

I can see that the site uses the official cometd implementation and it indeed seems to do even the first handshake with the websocket transport.
After some investigation in the source of the official clients I found that they do the transport negotiation step differently then how I interpreted the specifications.
Based on the specs I thought that the long-polling transport should be always active on the server, and based on the result of the first handshake, which contains a list of all the supported transport types on the server side, the client can chose a different transport type for subsequent requests.
It turns out that the official clients will instead use a trial and error method, they will try to handshake with all the transport types supported by the client, one by one in some predefined order, until they're able to make a successful handshake.
Based on this discovery, I will have to reorganize the library in the future and move some of the logic implemented in the transports to the client class. But I think this will take a longer amount of time...
In the meantime, until the mentioned changes are implemented, you can do the following simple trick to make the client work with websocket only connections:

import asyncio

from aiocometd import Client, ConnectionType
import aiocometd.client


async def main():
    aiocometd.client.DEFAULT_CONNECTION_TYPE = ConnectionType.WEBSOCKET

    async with Client("https://tools.dxfeed.com/webservice/cometd") as client:
        async for message in client:
            print(message)

if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())

To solve or test your other issues, I would need to trace what the official client does on https://tools.dxfeed.com/webservice/debug-console.jsp
My problem is that I'm totally unfamiliar with this domain. This might sound funny, but can you tell me step-by-step what buttons should I press or what text to enter in which text box on https://tools.dxfeed.com/webservice/debug-console.jsp to get the output from your screenshot? 😄

[boyan-soubachov]

Thank you for the information. That's very useful for me to know.

I'd be very happy to contribute a PR to aiocometd if you're accepting them? For the test client on the site (https://tools.dxfeed.com/webservice/debug-console.jsp) just do the following steps:

1. Right click and use the "Inspect" tool to view the network connections going in and out
2. Refresh the page
3. Tick & enable the "Profile" option (first one in Market events)
4. Click Create subscription at the bottom of the page
5. A new textbox will be created at the bottom of the page; in the "Symbol" field, just type /ES and hit Enter.

You'll see in your Network inspector that there will be a websocket connection made

[robertmrk]

@boyan-soubachov
Thank you for the directions!
I found the problem.

So the specifications allow 2 simultaneous connections for HTTP based transports. In fact, 2 connections are the minimum for them to be able to use them in full duplex mode. While for non HTTP transports like the websocket it's not exactly clear if it's allowed or not. When I implemented the websocket transport I noticed that the cometd example applications (which I use for integration testing) allow 2 simultaneous websocket connections as well. This allowed me to use a much simpler implementation than what I would ended up with using a single connection. Obviously I went for the simpler one.
The service implementation at dxfeed for some reason doesn't allows more then a single connection per client. So when we try to publish something after a successful connection, it responds with 402 on the second connection, (which is weird in it itself because that response is reserved for cases when the client forgot to do a handshake or if there were a long network outage).
I'm not sure why their service acts differently compared to the demo applications. I suspect there is a configuration option on the server side for this.

TL;DR; The websocket transport needs to be implemented using a single connection to solve this problem.

Normally I would be more than happy to accept PRs, but I plan to implement this myself, since I know how tricky it is to implement the automatic reconnection, and network outage detection features of the library with websockets. But it would be very useful if you could help me with the testing of this fix with a real world workload. I'll soon create a new feature branch for testing this.

[boyan-soubachov]

Thank you, @robertmrk !

This is very useful to know. I'm happy to test the proposed change/branch. I will modify my code to set the DEFAULT_CONNECTION_TYPE to websockets as you mentioned earlier and will try with your new, single-connection websocket mode when ready.

Do you plan to implement this as a parameter/option or do you want to make it the default. I'm not too familiar with the Bayeux protocol specification but from a very quick guess, it seems to me like a better idea to make 1 connection the default since it would be more widely supported given what you say?

[robertmrk]

Hi @boyan-soubachov

Please try connect to dxfeed with the implementation from the branch feature/single-connection-websocket.
You can install it like this:

$ pip install -e git+https://github.com/robertmrk/aiocometd.git@feature/single-connection-websocket#egg=aiocometd

If it works, then this will be the new implementation for websocket transport. There is no point for maintaining two implementations for it, there should be no significant difference in performance.

[boyan-soubachov]

That seems to have done the trick! I've got some major fixes to do on my side but everything looks good for now. Thank you!

I will let you know if I stumble across anything in the future but will switch over to aiocometd when this makes it into master. Thanks again!

[robertmrk]

You're welcome!
The changes have been merged and you can get them by installing the latest release.
I'll close this issue for now since it looks like the new websocket transport implementation fixed it. If you encounter any new problems related to this then please reopen this issue.