Websockets Feed

[AusIV]

This is an older ticket, but it looks like it may be finally getting some legs, so I'm writing it up in more detail.

We need a microservice that implements the Websockets version of the Standard Relayer API.

The microservice will subscribe to a list of channels. Whenever an order comes in on any of the channels, the microservice will iterate over its list of websocket subscribers, check whether the order matches the filters of each subscriber, and send the order to the subscriber if it matches.

The websocket portion of the service should probably use Gorilla Websockets to upgrade HTTP connections into Websockets Connections. When a user submits a "subscribe" message as laid out in the Standard Relayer API, it should create a subscription object that includes.

Subscription Object

The Subscription object should include:

• The request ID
• A filter object
• A chan *types.Order channel, which should be sent orders that match the filter.
• A boolean indicating whether the channel is still open.

Subscriptions should have a Dispatch(*types.Order) method, which will be called on all available orders. The dispatch method should evaluate the received order against the filter object. If the object matches the filter and the connection is still open, the order should be sent over the subscription's channel

Filter Object

The filter object should have attributes for:

• makerAssetProxyId ([4]byte)
• takerAssetProxyId ([4]byte)
• makerAssetAddress (*types.Address)
• takerAssetAddress (*types.Address)
• makerAssetData (*types.AssetData)
• takerAssetData (*types.AssetData)
• traderAssetData (*types.AssetData)
• networkId (int)

It should include a Filter(*types.Order) method. Which should return true if the message should be forwarded, false otherwise. The filter method should check each property of the filter. If the filter property is null, then no checks are necessary. If the filter property is non-null, it should be compared against the corresponding order property or properties. If the properties match, it should continue to the next property, otherwise it should return false.

Websocket Goroutine

When a user connects via websockets, the HTTP process will instantiate a new Goroutine for that user and call the handler function. Our handler function should upgrade the connection to a websockets request. After upgrading, the handler function should:

1. Create an order channel

2. Create a go routine to watch that order channel

3. When orders come in through that order channel, serialized them based on github.com/notegio/openrelay/search.GetFormattedOrder() and sent over the channel.

   GetFormattedOrder() takes a *db.Order rather than a *types.Order. For this, we can construct a *db.Order given the *types.Order(see github.com/notegio/openrelay/db/indexer.go - index function), then call order.Populate() to fill in the other necessary details.

4. Read for new messages on the websockets connection. If it gets a subscription message, it should:

   1. Construct a subscription object based on the incoming message
   2. Add the new subscription object to a global list of subscriptions (iterate over the list of subscriptions and replace the first subscription where the "active" boolean field for the subscription is False, or append a new one to the end if none are inactive).
   3. Spin off a new goroutine to watch the channel for that subscription
   4. In the new goroutine, when messages come in on the subscription channel, simply relay them to the websocket's order channel (it needs to go through that channel because only one goroutine at a time is allowed to write to the socket, so we'll just pass messages to that goroutine).

5. If a message comes in indicating that the socket has closed, or if a write operation to the websocket fails, mark the subscription as inactive.

Consumer

Each channel we listen on will require a Consumer implementing the github.com/notegio/openrelay/channels/consumer.Consumer interface. The Consume(Delivery) method will be called each time an order is received. The consumer should iterate over the list of subscriptions, calling Dispatch on each. These should be added to ConsumerChannels, following the same pattern as most of the other OpenRelay consumers.

Command

The structure of the command for this microservice will be comparable to that of cmd/searchapi/main.go. It will have one API endpoint, a list of Consumers. The command line arguments will consist of the redis URL and a list of channels to watch. The channels should be converted to ConsumerChannels using the channels.ConsumerFromURI method, allowing it to operate from both Topic and Queue based channels.

Testing

This service can be tested in several smaller units:

Filters can simply be tested by checking whether a filter matches a provided order as expected.

Subscriptions can be tested by passing calling Dispatch() and testing whether the order goes on the channel as expected.

Testing the websockets piece will require a websockets client implemented in Go that sends the appropriate subscription messages, and watches for orders to come back as expected. Sending orders to the consumers that will dispatch through the websockets can be done with the MockChannel library.