using System;

using System.Text.Json;

using System.Threading.Tasks;

using System.Collections.Generic;

using Microsoft.Extensions.DependencyInjection;

public class Program

{

    public static async Task Main()

    {

        var container = BootstrapContainer();

​

        var svc = container.GetService<OfframpQueueService>();

        await svc.RunAsync();

    }

    public static IServiceProvider BootstrapContainer()

    {

        return  new ServiceCollection()

            .AddSingleton<IOperationHandler<OfframpMessage1>>(new OperationHandler<OfframpMessage1>(new OfframpMessage1Handler()))

            .AddSingleton<IOperationHandler<OfframpMessage2>>(new OperationHandler<OfframpMessage2>(new OfframpMessage2Handler()))

            .AddSingleton<IQueue<OfframpMessage>>(new GenericQueue<OfframpMessage>(new OfframpQueue()))

            .AddSingleton<OfframpQueueService>(sp => {

                var processors = new Dictionary<string,IOperationHandler> 

                {

                    ["Op1"] = sp.GetService<IOperationHandler<OfframpMessage1>>(),

                    ["Op2"] = sp.GetService<IOperationHandler<OfframpMessage2>>()

                };

                return new OfframpQueueService(processors, sp.GetService<IQueue<OfframpMessage>>());

            })

            .BuildServiceProvider();

    }

}

​

public interface IOperationHandler<T> : IOperationHandler { }

public interface IOperationHandler

{

    void Handle(Message message);

}

​

public class OperationHandler<T> : IOperationHandler<T>

{

    private readonly IOperationHandler _handler;

    public OperationHandler(IOperationHandler handler)

    {

        _handler = handler;

    }

​

    public virtual void Handle(Message message)

    {

        _handler.Handle(message);

    }

}

​

public class OfframpMessage1Handler : IOperationHandler

{

    public virtual void Handle(Message message)

    {

        Console.WriteLine($"Message1Handler, handling {message.PayloadJson}");

    }

}

​

public class OfframpMessage2Handler : IOperationHandler

{

    public virtual void Handle(Message message)

    {

        Console.WriteLine($"Message2Handler, handling {message.PayloadJson}");

    }

}

​

public interface IQueue<T> : IQueue { }

​

public interface IQueue

{

        Task<bool> TryGetMessageAsync(out Message message);

}

public class GenericQueue<T> : IQueue<T>

{

    private readonly IQueue _queue;

    public GenericQueue(IQueue queue)

    {

        _queue = queue;

    }

    public virtual Task<bool> TryGetMessageAsync(out Message message)

    {

        return _queue.TryGetMessageAsync(out message);

    }

}

​

public class Message

{

    public string RawValue { get; set; }

    public string Operation { get; set; }

    public string PayloadJson { get; set; }

}

​

public class OfframpMessage : Message

{

}

public class OfframpMessage1 : OfframpMessage

{

}

​

public class OfframpMessage2 : OfframpMessage

{

}

​

public class OfframpQueue: IQueue<OfframpMessage>

{

    private readonly Queue<string> _queue = new Queue<string>();

​

    public OfframpQueue()

    {

        ResetQueue();

    }

    private void ResetQueue()

    {

        var context = new System.Text.Json.JsonSerializerOptions { AllowTrailingCommas = true, PropertyNamingPolicy = JsonNamingPolicy.CamelCase };

        _queue.Clear();

        _queue.Enqueue(JsonSerializer.Serialize(new { Operation = "Op1", Payload = new { CorrelationId = Guid.NewGuid(), TriggerValue = 0.80m }}, context));

        _queue.Enqueue(JsonSerializer.Serialize(new { Operation = "Op2", Payload = new { CorrelationId = Guid.NewGuid(), AdditionalData = "Hoo", Reason = "scanned" }}, context));

    }

​

    public Task<bool> TryGetMessageAsync(out Message message)

    {

        var hasMessage = _queue.TryDequeue(out var msg);

        message = new Message { RawValue = msg };

        if (!string.IsNullOrWhiteSpace(message.RawValue))

        {

            var jsonDoc = JsonDocument.Parse(message.RawValue);

            message.Operation = jsonDoc.RootElement.GetProperty("operation").GetString();

            message.PayloadJson = jsonDoc.RootElement.GetProperty("payload").ToString();

        }

        return Task.FromResult(hasMessage);

    }

}

​

public class OfframpQueueService

{

    public const int MAX_DELAY_UNITS = 16;

    private readonly Dictionary<string, IOperationHandler> _handlers;

    private readonly IQueue _inputQueue;

    private int _delayUnit = 1;

    public OfframpQueueService(Dictionary<string, IOperationHandler> handlers, IQueue<OfframpMessage> inputQueue)

    {

        _handlers = handlers;

        _inputQueue = inputQueue;

    }

    public async Task RunAsync()

    {

        var count = 0;

        while (count++ < 5) //this is just a hack to give this thing 5 tries, can't run it indefinitely on .net fiddle, in real world, this is while(true)

        {

            if (!await _inputQueue.TryGetMessageAsync(out var message))

            {

                await ExecuteExponentialBackOff();

                continue;

            }

            Console.WriteLine($"Processing message {message.RawValue}");

            var opHandler = FindOperationHandlerFor(message.Operation);

            opHandler.Handle(message);

        }       

    }

    private Task ExecuteExponentialBackOff()

    {

        Console.WriteLine($"Backing off for {_delayUnit * 100} millisecond(s)...");

        var currentDelayUnit = _delayUnit;

        _delayUnit = _delayUnit < MAX_DELAY_UNITS ? _delayUnit * 2 : MAX_DELAY_UNITS;

        return Task.Delay(TimeSpan.FromMilliseconds(currentDelayUnit * 100));

    }

    private IOperationHandler FindOperationHandlerFor(string operation)

    {

        if (!_handlers.ContainsKey(operation)) throw new ArgumentOutOfRangeException(nameof(operation));

        return _handlers[operation];

    }

}

​