Return to Revisions

Even though you have not explicitly mentioned it in your question but I assume you are using this library.

Your SubscribeToAsync is a good starting point. But your to be provided func parameters are still have much resemblance as you can see

await SubscribeToAsync(() => _socketClient.Streams.SubscribeToOrderUpdatesAsync(data =>
{
    _logger.LogInformation("Orders: {@Data}", data.Data);
}, cancellationToken));

await SubscribeToAsync(() => _socketClient.Streams.SubscribeToUserTradeUpdatesAsync(data =>
{
    _logger.LogInformation("Fills: {@Data}", data.Data);
}, cancellationToken));

If you want to move as much logic as possible into your SubscribeToAsync then you have to assess the above method calls. What are the differences?

• SubscribeToOrderUpdatesAsync vs SubscribeToUserTradeUpdatesAsync
• "Orders: vs "Fills:

If you could provide these information to the SubscribeToAsync methods then you could get rid of the SubscribeToOrdersAsync and SubscribeToFillsAsync methods entirely.

Fortunately you can do that:

private async Task<UpdateSubscription?> SubscribeToAsync<T>(
    Func<IFTXSocketClientStreams, Func<Action<DataEvent<T>>, CancellationToken, Task<CallResult<UpdateSubscription>>>> methodSelector,
    string logPrefix, CancellationToken token)
{
    var asyncMethod = methodSelector(_socketClient.Streams);
    var subscription = await asyncMethod(data => _logger.LogInformation(logPrefix +": {@Data}", data.Data), token);

    if (!subscription.Success)
        return null;

    subscription.Data.ConnectionLost += () => _logger.LogError("Connection lost");
    subscription.Data.ConnectionRestored += _ => _logger.LogInformation("Connection restored");

    return subscription.Data;
}

• The methodSelector parameter: It select the to be called method from the IFTXSocketClientStreams inteface
  • Since FTXOrder and FTXUserTrade don't have a common base class that's why you have to make the SubscribeToAsync generic
• The logPrefix parameter: The prefix of the logging message
• The token parameter: The cancellation token

With this in your hand the entire class could be shortened like this:

public class FtxProducer : IHostedService
{
    private readonly ILogger<FtxProducer> _logger;
    private readonly IFTXSocketClient _socketClient;

    public FtxProducer(ILogger<FtxProducer> logger, IFTXSocketClient socketClient)
        => (_logger, _socketClient) = (logger, socketClient);

    public async Task StartAsync(CancellationToken cancellationToken)
    {
        await SubscribeToAsync<FTXOrder>(stream => stream.SubscribeToOrderUpdatesAsync, "Orders", cancellationToken);
        await SubscribeToAsync<FTXUserTrade>(stream => stream.SubscribeToUserTradeUpdatesAsync, "Fills", cancellationToken);
    }

    public Task StopAsync(CancellationToken cancellationToken)
        => _socketClient.UnsubscribeAllAsync();

    private async Task<UpdateSubscription?> SubscribeToAsync<T>(
        Func<IFTXSocketClientStreams, Func<Action<DataEvent<T>>, CancellationToken, Task<CallResult<UpdateSubscription>>>> methodSelector,
        string logPrefix, CancellationToken token)
    {
        var asyncMethod = methodSelector(_socketClient.Streams);
        var subscription = await asyncMethod(data => _logger.LogInformation(logPrefix +": {@Data}", data.Data), token);

        if (!subscription.Success) return null;

        subscription.Data.ConnectionLost += () => _logger.LogError("Connection lost");
        subscription.Data.ConnectionRestored += _ => _logger.LogInformation("Connection restored");

        return subscription.Data;
    }
}

UPDATE #1

May you just change _logger.LogInformation(logPrefix +": {@Data}", data.Data) to something more generic?

private async Task<UpdateSubscription?> SubscribeToWithLoggingAsync<T>(
    Func<IFTXSocketClientStreams, Func<Action<DataEvent<T>>, CancellationToken, Task<CallResult<UpdateSubscription>>>> methodSelector,
    string logPrefix, CancellationToken token)
    => await SubscribeToAsync(methodSelector, data => _logger.LogInformation(logPrefix +": {@Data}", data.Data), token);

private async Task<UpdateSubscription?> SubscribeToAsync<T>(
    Func<IFTXSocketClientStreams, Func<Action<DataEvent<T>>, CancellationToken, Task<CallResult<UpdateSubscription>>>> methodSelector,
    Action<DataEvent<T>> handler, CancellationToken token)
{
    var asyncMethod = methodSelector(_socketClient.Streams);
    var subscription = await asyncMethod(handler, token);

    if (!subscription.Success) return null;

    subscription.Data.ConnectionLost += () => _logger.LogError("Connection lost");
    subscription.Data.ConnectionRestored += _ => _logger.LogInformation("Connection restored");

    return subscription.Data;
}

• The SubscribeToWithLoggingAsync is a specialized version of the more generic SubscribeToAsync
  • It specifies that the handler should perform some simple logging
• The SubscribeToAsync method's signature looks a bit terrifying now, but it is generic enough to be reused in multiple ways

UPDATE #2

public abstract Task<CallResult<UpdateSubscription>> SubscribeToTickerUpdatesAsync(string symbol, Action<DataEvent<FTXStreamTicker>> handler, CancellationToken ct = default(CancellationToken)) I wasn't able to call this method.

If you look at the signature of the method then you can see there is an extra input parameter called symbol. The SubscribeToAsync's methodSelector is unaware of this.

In order to support the call of SubscribeToTickerUpdatesAsync or SubscribeToTradeUpdatesAsync like this:

await SubscribeWithSymbolToAsync<FTXStreamTicker>(stream => stream.SubscribeToTickerUpdatesAsync, "Ticker", cancellationToken);
await SubscribeWithSymbolToAsync<IEnumerable<FTXTrade>>(stream => stream.SubscribeToTradeUpdatesAsync, "Trade", cancellationToken);

we need the following trick:

private async Task<UpdateSubscription?> SubscribeWithSymbolToAsync<T>(
    Func<IFTXSocketClientStreams, Func<string, Action<DataEvent<T>>, CancellationToken, Task<CallResult<UpdateSubscription>>>> methodSelector,
    string symbol, CancellationToken token)
{
    var asyncMethod = (Action<DataEvent<T>> handler, CancellationToken cToken) =>
        methodSelector(_socketClient.Streams)(symbol, handler, cToken);

    var subscription = await asyncMethod(data => _logger.LogInformation(symbol + ": {@Data}", data.Data), token);

    return SubscribeToEvents(subscription);
}

private async Task<UpdateSubscription?> SubscribeToAsync<T>(
    Func<IFTXSocketClientStreams, Func<Action<DataEvent<T>>, CancellationToken, Task<CallResult<UpdateSubscription>>>> methodSelector,
    Action<DataEvent<T>> handler, CancellationToken token)
{
    var asyncMethod = methodSelector(_socketClient.Streams);
    var subscription = await asyncMethod(handler, token);

    return SubscribeToEvents(subscription);
}

private UpdateSubscription SubscribeToEvents(CallResult<UpdateSubscription> subscription)
{
    if (!subscription.Success) return null;

    subscription.Data.ConnectionLost += () => _logger.LogError("Connection lost");
    subscription.Data.ConnectionRestored += _ => _logger.LogInformation("Connection restored");

    return subscription.Data;
}

• The SubscribeWithSymbolToAsync binds the symbol formal parameter of the methodSelector to the provided symbol parameter
  • In other words, we convert a Func<T1, T2, T3> to a Func<T2, T3> by supplying T1
• The SubscribeWithSymbolToAsync does not rely on SubscribeToAsync so I extract the common part
  • The SubscribeToEvents handles the event subscribtions