I want to hold some informations when some execution is running like CorrelationId, I tried to create a static class with an AsyncLocal object like this:
public static class InternalContext
{
private static readonly AsyncLocal<Dictionary<string, string>> InternalContextHolder =
new AsyncLocal<Dictionary<string, string>>() { Value = new Dictionary<string, string>() };
private static Dictionary<string, string> ContextHolder => InternalContextHolder.Value ??= new Dictionary<string, string>();
public const string CORRELATION_ID_KEY = "correlationId";
public static string CorrelationId
{
get
{
if (!ContextHolder.TryGetValue(key, out string currentValue))
{
ContextHolder[CORRELATION_ID_KEY] = Guid.NewGuid().ToString();
}
return ContextHolder[CORRELATION_ID_KEY];
}
set
{
ContextHolder[CORRELATION_ID_KEY] = value;
}
}
public static void ResetContext()
{
CorrelationId = null;
ContextHolder.Clear();
}
}
This is my code that call AWS SQS and get the messages in the queue:
public async Task<IEnumerable<QueueMessage>> GetMessages() {
var receiveMessageRequest = new ReceiveMessageRequest
{
AttributeNames = new List<string> { "All" },
MessageAttributeNames = new List<string>() { "All" },
MaxNumberOfMessages = 1,
QueueUrl = await amazonSQS.GetQueueUrlAsync("my-test-queue")
};
var result = await amazonSQS.ReceiveMessageAsync(receiveMessageRequest);
// Convert result to internal QueueMessage object
return result.ToQueueMessageList();
}
public string GetMessage()
{
var message = GetMessages().FirstOrDefault();
if (message != null)
{
InternalContext.CorrelationId = message.MessageAttributes["correlationId"];
}
return message?.body;
}
Then I have a BackgroundService running with this:
public virtual async Task ExecuteAsync(CancellationToken stoppingToken)
{
var msg = GetMessage();
// Couple HttpClient calls that is injecting the InternalContext.CorrelationId in the request headers for propagation
// In the end of execution, reset the context to avoid "cached" values
InternalContext.ResetContext();
}
So basically it works fine with 1 message in the queue, the biggest problem is that I'm now trying to get multiple messages from the SQS to make things faster, but when I try to process them in parallel the InternalContext get lost, using the same context instance for multiple Tasks, like this:
public virtual async Task ExecuteAsync(CancellationToken stoppingToken)
{
IEnumerable<QueueMessage> messagesList = await GetMessages();
var tasks = messagesList.Select(msg => Task.Run(async () =>
{
InternalContext.CorrelationId = msg.MessageAttributes["correlationId"];
// Couple HttpClient calls that is injecting the InternalContext.CorrelationId in the request headers for propagation
// In the end of execution, reset the context to avoid "cached" values
InternalContext.ResetContext();
}));
await Task.WhenAll(tasks);
}
But in this way above the InternalContext is conflicting with the Tasks, so it's like the first 3~5 messages send the CorrelationId from the first message, the messages 6~10 send the CorrelationId from the message X (they are "random").
So my guess is that this Context I create is not the best way to do it, but I have no idea how to deal with it.
The propose of this Context is to store data like the correlation id, trace id, and any other value I need, to send it to any request I make, doesn't matter if it's a HTTP call using HttpClient, or if I'm sending some message to a new Queue, or to a SNS topic, I kinda need to hold it's value while my code is executing and that it does't get lost with multi threading (which I tought AsyncLocal would help me with).