using System;

using System.Collections.Generic;

using System.Diagnostics.CodeAnalysis;

using System.Reactive.Concurrency;

using System.Reactive.Linq;

using System.Reactive.Subjects;

using System.Threading.Tasks;

​

var result = await Test();

​

if (result.Success)

{

    foreach (var item in result.Data)

    {

        Console.WriteLine(item);

    }

}

else

{

    Console.WriteLine($"Error {result.Error.Message}");

}

​

static async Task<Result<IEnumerable<Details>>> Test()

{

    var client = new IBClient();

    var requestId = 1;

    var results = new ReplaySubject<Details>();

​

    var endSignal = client.DetailsEnd

        .Any(x => x.RequestId == requestId);

    // TODO: Return "Result<IEnumerable<Details>>.FromError(new Error(123, "asd", null));" in case of timeout or error.

    var errorSignal = client.Error

        .Where(x => x.RequestId == requestId)

        .Subscribe(x => results.OnError(new Exception($"Code: {x.Code}, Message: {x.Message}")));

​

    using (client.Details

               .Where(x => x.RequestId == requestId)

               .TakeUntil(endSignal)

               .Timeout(TimeSpan.FromSeconds(2))

               .Catch<Details, TimeoutException>(ex =>

               {

                   Console.WriteLine("Called");

                   return Observable.Empty<Details>();

               })

               .ObserveOn(TaskPoolScheduler.Default)

               .Subscribe(results))

    {

        client.Emit(2, 4);

        client.Emit(requestId, 42);

        client.EmitEnd(2);

        client.Emit(requestId, 62);

        client.Emit(requestId, 123);

        client.EmitEnd(requestId);

        client.Emit(requestId, 1);

​

        // Expected output: 42, 62, 123.

        await results;

​

        var items = results.ToEnumerable();

        return Result<IEnumerable<Details>>.FromSuccess(items);

    }

}

​

public record Details(int RequestId, int Value);

​

public record DetailsEnd(int RequestId);

​

public record ErrorHappened(int RequestId, int Code, string Message);

​

public class IBClient

{

    private readonly Subject<ErrorHappened> _errorSubject = new();

    public IObservable<ErrorHappened> Error => _errorSubject.AsObservable();

​

    private readonly Subject<Details> _subjectDetailsSubject = new();

    public IObservable<Details> Details => _subjectDetailsSubject.AsObservable();

    private readonly Subject<DetailsEnd> _subjectDetailsEndSubject = new();

    public IObservable<DetailsEnd> DetailsEnd => _subjectDetailsEndSubject.AsObservable();

​

    public void Emit(int requestId, int value) => _subjectDetailsSubject.OnNext(new Details(requestId, value));

​

    public void EmitEnd(int requestId) => _subjectDetailsEndSubject.OnNext(new DetailsEnd(requestId));

​

    public void EmitError(int requestId, int code, string message) => _errorSubject.OnNext(new ErrorHappened(requestId, code, message));

}

​

public record Error(int? Code, string Message, object? Data);

​

public record Result<T>(

    [property: MemberNotNullWhen(true, nameof(Result<T>.Data))]

    [property: MemberNotNullWhen(false, nameof(Result<T>.Error))]

    bool Success, T? Data, Error? Error)

{

    public static Result<T> FromSuccess(T data) => new(true, data, default);

    public static Result<T> FromError<TError>(TError error) where TError : Error => new(false, default, error);

}