using System;

using System.Text.Json;

using System.Threading;

using System.Threading.Tasks;

using System.Collections.Generic;

public class Program

{

    public static async Task Main()

    {

        var queue = new FakeQueue();

        //Setup the queue handler

        var processors = new Dictionary<string, IDeepQueueMessageProcessor> 

        {

            [DeepQueueOperation.Foo] = new FooOperationDeepQueueMessageProcessor(),

            [DeepQueueOperation.Bar] = new BarOperationDeepQueueMessageProcessor()

        };

        var queueHandler = new FooBarDeepQueueMessageHandler(processors);

        //Process mesage from the queue

        var cts = new CancellationTokenSource();

        while (queue.TryGetMessage(out var message))

        {

            Console.WriteLine($"Got a message! {message}");

            var context = new DeepQueueContext(message);        

            context = await queueHandler.HandleAsync(context, cts.Token);

            //Do whatever you want with the context here, like process context.Response, etc.

            Console.WriteLine($"Message was handled? {context.Response.IsHandled}");

        }

        Console.WriteLine("Done");

    }

}

​

public class FakeQueue

{

    private Queue<string> _messages = new Queue<string>();

    public FakeQueue()

    {

        ResetQueue();

    }

    public void ResetQueue()

    {

        //This example ties the message directly to Operation which is not ideal...

        //You can convert this into a state machine message where you provide the state and then drive the operation from a

        //translation table that can be dynamically loaded via storage, etc. that map the route between state and target Operation

        //When state is A, then the next operation is B which allow the operation to determine what it next state is going to be,

        //potentially, each operation much have multiple outcome and some might need to be forward to a certain and some will not.

        var message1 = new

        {

            Operation = DeepQueueOperation.Foo, //This is a MUST have, since we will use this to determine the processor to use...

            Payload = new                       //This is a MUST have and it will contain arbitrary nodes

            {                                   //This section can be anything and custom to each processor...

                CorrelationId = Guid.NewGuid(),

                FooData = "Whatever!"

            }

        };

        var message2 = new

        {

            Operation = DeepQueueOperation.Bar, //This is a MUST have, since we will use this to determine the processor to use...

            Payload = new                       //This is a MUST have and it will contain arbitrary nodes

            {                                   //This section can be anything and custom to each processor...

                BarNum = 11050

            }

        };      

​

        var message3 = new

        {

            Operation = DeepQueueOperation.Bar, //This is a MUST have, since we will use this to determine the processor to use...

            Payload = new                       //This is a MUST have and it will contain arbitrary nodes

            {                                   //This section can be anything and custom to each processor...

                BarNum = -1000

            }

        };      

​

        var message4 = new

        {

            Operation = DeepQueueOperation.Foo, //This is a MUST have, since we will use this to determine the processor to use...

            Payload = new                       //This is a MUST have and it will contain arbitrary nodes

            {                                   //This section can be anything and custom to each processor...

                CorrelationId = Guid.NewGuid(),

                FooData = ""

            }

        };

​

        _messages.Clear();

        _messages.Enqueue(JsonSerializer.Serialize(message1));

        _messages.Enqueue(JsonSerializer.Serialize(message2));

        _messages.Enqueue(JsonSerializer.Serialize(message3));

        _messages.Enqueue(JsonSerializer.Serialize(message4));

    }

    public bool TryGetMessage(out string message)

    {

        return _messages.TryDequeue(out message);

    }

}

​

public interface IDeepQueueMessageHandler

{

    Task<DeepQueueContext> HandleAsync(DeepQueueContext context, CancellationToken token = default);    

}

​

public class FooBarDeepQueueMessageHandler : IDeepQueueMessageHandler

{

    private readonly Dictionary<string, IDeepQueueMessageProcessor> _processors;

    public FooBarDeepQueueMessageHandler(Dictionary<string, IDeepQueueMessageProcessor> processors)

    {

        _processors = processors;

    }

    public async Task<DeepQueueContext> HandleAsync(DeepQueueContext context, CancellationToken token)

    {

        context = context.PrepareForProcessing();

        var processor = FindProcessorForOperation(context.Operation);

        context = processor.EnsureContextRequestFor(context);

        context.Response = await processor.ProcessAsync(context.Request);

        return context;

    }

    private IDeepQueueMessageProcessor FindProcessorForOperation(string operation)

    {

        if (string.IsNullOrWhiteSpace(operation)) throw new ArgumentNullException(nameof(operation));

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

        return _processors[operation];

    }

}

​

public class DeepQueueContext

{

    public IDeepQueueOperationRequest Request { get; set; }

    public IDeepQueueOperationResponse Response { get; set; }

    public string RawJsonMessage { get; private set; }

    public string Operation { get; private set; }

    public string PayloadJson { get; private set; }

    public DeepQueueContext(string rawMessageJson) 

    {

        RawJsonMessage = rawMessageJson;

    }

    public DeepQueueContext PrepareForProcessing()

    {   

        var jsonDoc = JsonDocument.Parse(RawJsonMessage, new JsonDocumentOptions { AllowTrailingCommas = true });

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

        PayloadJson = jsonDoc.RootElement.GetProperty("Payload").ToString();        

        return this;

    }

}

​

public interface IDeepQueueMessageProcessor

{

    string OperationName { get; }

    Task<IDeepQueueOperationResponse> ProcessAsync(IDeepQueueOperationRequest request, CancellationToken token = default);

    DeepQueueContext EnsureContextRequestFor(DeepQueueContext context);

}

​

public interface IDeepQueueOperationRequest

{

}

​

public interface IDeepQueueOperationResponse

{

    bool IsHandled { get; set; }

}

​

public class FooDeepQueueOperationRequest : IDeepQueueOperationRequest

{

    public string Data { get; private set; }

​

    public FooDeepQueueOperationRequest(string data)

    {

        Data = data;

    }

    public override string ToString()

    {

        return JsonSerializer.Serialize(this);

    }

}

​

public abstract class DeepQueueOperationResponseBase : IDeepQueueOperationResponse

{

    public bool IsHandled { get; set; }

    public DeepQueueOperationResponseBase()

    {

        IsHandled = false;

    }

}

​

public class FooDeepQueueOperationResponse : DeepQueueOperationResponseBase

{   

    public FooDeepQueueOperationResponse(bool handled)

    {

        IsHandled = handled;

    }

    public FooDeepQueueOperationResponse() : this(false)

    {

    }

}

​

public class BarDeepQueueOperationRequest : IDeepQueueOperationRequest

{

    public long Number { get; private set; }

    public BarDeepQueueOperationRequest(long number)

    {

        Number = number;

    }

​

    public override string ToString()

    {

        return JsonSerializer.Serialize(this);

    }

}

​

public class BarDeepQueueOperationResponse : DeepQueueOperationResponseBase

{

    public BarDeepQueueOperationResponse(bool handled)

    {

        IsHandled = handled;

    }

​

    public BarDeepQueueOperationResponse() : this(false)

    {

    }

}

​

​

public class FooOperationDeepQueueMessageProcessor : IDeepQueueMessageProcessor

{

    public string OperationName => DeepQueueOperation.Foo;

    public async Task<IDeepQueueOperationResponse> ProcessAsync(IDeepQueueOperationRequest request, CancellationToken token)

    {

        var fooRequest = request as FooDeepQueueOperationRequest;

        Console.WriteLine($"Processing Foo Operation for {fooRequest}");

        await Task.Delay(1000);

        Console.WriteLine($"The foo data is \"{fooRequest.Data}\"");

​

        //We'll only return as handled when the data string has value...

        return new FooDeepQueueOperationResponse(!string.IsNullOrWhiteSpace(fooRequest.Data));

    }

    public DeepQueueContext EnsureContextRequestFor(DeepQueueContext context)

    {

        var jsonDoc = JsonDocument.Parse(context.PayloadJson, new JsonDocumentOptions { AllowTrailingCommas = true });

        var fooData = jsonDoc.RootElement.GetProperty("FooData").GetString();

        context.Request = new FooDeepQueueOperationRequest(fooData);

        return context;

    }

}

​

public class BarOperationDeepQueueMessageProcessor : IDeepQueueMessageProcessor

{

    public string OperationName => DeepQueueOperation.Bar;

    public async Task<IDeepQueueOperationResponse> ProcessAsync(IDeepQueueOperationRequest request, CancellationToken token)

    {

        var barRequest = request as BarDeepQueueOperationRequest;

        Console.WriteLine($"Processing Bar Operation for {barRequest}");

        await Task.Delay(1000);

        Console.WriteLine($"The bar number is {barRequest.Number}");

        //We'll only return as handled when the number is > -1...

        return new BarDeepQueueOperationResponse(barRequest.Number > -1);

    }

    public DeepQueueContext EnsureContextRequestFor(DeepQueueContext context)

    {

        var jsonDoc = JsonDocument.Parse(context.PayloadJson, new JsonDocumentOptions { AllowTrailingCommas = true });

        var number = jsonDoc.RootElement.GetProperty("BarNum").GetInt64();

        context.Request = new BarDeepQueueOperationRequest(number);

        return context;

    }

}

​

public static class DeepQueueOperation

{

    public const string Foo = "ApplyFoo";

    public const string Bar = "ApplyBar";

}