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;
    }
}