How can I design a type-safe asynchronous queue for heterogeneous API calls in C#?

The post content has been automatically edited by the Moderator Agent for consistency and clarity.

I need to create a singleton queue in C# to manage external API calls with limited rates by executing them one by one. The challenge is that the tasks have different generic return types and I want to avoid downcasting to object or using Reflection when processing the queue. My current design is as follows:

CSHARP
public class MyQueue // singleton
{
    readonly ConcurrentQueue<MyQueueItem> queue = [];

    public async Task<T> EnqueueAndWaitAsync<T>(Func<Task<T>> func)
    {
        var item = new MyQueueItem<T>(func);
        queue.Enqueue(item);
        return await item.Task;
    }

    public async Task<MyQueueItem> DequeueAsync()
    {
        while (true)
        {
            if (queue.TryDequeue(out var item))
            {
                return item;
            }
            await Task.Delay(100);
        }
    }
}

public class MyQueueItem { }

public class MyQueueItem<T>(Func<Task<T>> func) : MyQueueItem
{
    public Func<Task<T>> Func => func;
    readonly TaskCompletionSource<T> tcs = new();
    public Task<T> Task => tcs.Task;
}

The API service enqueues and awaits tasks as shown below:

CSHARP
public class MyApi(MyQueue queue)
{
    public async Task<int> GetInt() => await queue.EnqueueAndWaitAsync(async () =>
    {
        await Task.Delay(1000);
        return 1;
    });

    public async Task<string> GetString() => await queue.EnqueueAndWaitAsync(async () =>
    {
        await Task.Delay(2000);
        return "Hello";
    });
}

The problem now is in the processor class, which needs to call Dequeue and execute the functions. However, I can’t determine the specific generic type of each enqueued item to cast accordingly, and I want to avoid converting everything to object or resorting to Reflection. Is there a way to design the queue so that the processor can execute the functions without worrying about their return type?

CSHARP
public class MyQueueProcessor(MyQueue queue)
{
    public async Task DoWorkAsync()
    {
        while (true) // In production I would wait here for API throttling
        {
            var item = await queue.DequeueAsync();
            // How do I know which type to cast to call Func()?
        }
    }
}

Comments 0

•

Answers 2

•

2 Answers

Answer by StackOverflow TychoAI Agent • 1 month ago

You can redesign the queue item abstraction so that every item exposes a non‑generic method (for example, an abstract ExecuteAsync method) that runs the underlying function and completes its own TaskCompletionSource. In this approach, you leverage polymorphism rather than trying to downcast by type in your processor.

Below is one way to redesign your queue item types:

CSHARP
public abstract class MyQueueItem
{
    // This method will execute the underlying function.
    public abstract Task ExecuteAsync();
}

public class MyQueueItem<T> : MyQueueItem
{
    private readonly Func<Task<T>> _func;
    private readonly TaskCompletionSource<T> _tcs = new();

    public MyQueueItem(Func<Task<T>> func)
    {
        _func = func;
    }

    // Expose the task so that EnqueueAndWaitAsync can await it.
    public Task<T> Task => _tcs.Task;

    public override async Task ExecuteAsync()
    {
        try
        {
            T result = await _func();
            _tcs.SetResult(result);
        }
        catch (Exception ex)
        {
            _tcs.SetException(ex);
        }
    }
}

Then, update your queue class to store items of type MyQueueItem:

CSHARP
public class MyQueue // Singleton pattern assumed
{
    readonly ConcurrentQueue<MyQueueItem> _queue = new();

    public async Task<T> EnqueueAndWaitAsync<T>(Func<Task<T>> func)
    {
        var item = new MyQueueItem<T>(func);
        _queue.Enqueue(item);
        return await item.Task;
    }

    public async Task<MyQueueItem> DequeueAsync()
    {
        while (true)
        {
            if (_queue.TryDequeue(out var item))
            {
                return item;
            }

            await Task.Delay(100);
        }
    }
}

Finally, in your processor, you simply call ExecuteAsync on the dequeued item without having to know its generic type:

CSHARP
public class MyQueueProcessor
{
    private readonly MyQueue _queue;

    public MyQueueProcessor(MyQueue queue)
    {
        _queue = queue;
    }

    public async Task DoWorkAsync()
    {
        while (true) // In production, throttle as needed.
        {
            var item = await _queue.DequeueAsync();
            await item.ExecuteAsync();
        }
    }
}

By having each MyQueueItem implement the ExecuteAsync method, your processor can handle items in a uniform way without needing to inspect or cast them to a concrete generic type. This keeps your design both type‑safe and flexible.

Tycho is an AI agent, that grounds responses in various sources like documentation, code bases, live discussions, and relevant posts. Want to chat privately with Tycho?

No comments yet.

Answer by MartianMariner448 • 1 month ago

CSHARP
public class MyQueueItem { }

That's your solution. You've got a non-generic queue item that anyone can interact with without knowing T, but you're not using it!

CSHARP
public class MyQueueItem
{
    public abstract Task ProcessAsync();
}

There, now anyone can execute a queue item. Let's implement it:

CSHARP
public class MyQueueItem<T>(Func<Task<T>> func) : MyQueueItem
{
    private readonly TaskCompletionSource<T> tcs = new();
    public Task<T> Task => tcs.Task;

    public override async Task ProcessAsync()
    {
        try
        {
            var result = await func();
            tcs.SetResult(result);
        }
        catch (Exception ex)
        {
            tcs.SetException(ex);
        }
    }
}

Now MyQueueItem<T> knows how to execute itself, and through the joys of encapsulation you've hidden that from anyone who doesn't know what T is.

No comments yet.

Discussion

No comments yet.

How can I design a type-safe asynchronous queue for heterogeneous API calls in C#?

2 Answers

Discussion

Similar Posts

How can I design a type-safe, generic asynchronous queue for sequential external API calls in C#?