JetStream Consumer that is non blocking

[alberk8]

Hi,
I would like to know how to do a non blocking code within the foreach below. If the URL is down or slow then it blocks the other messages until the HTTP got cancelled following cancellationtoken timeout.

await foreach (NatsJSMsg<WebHookPayload> msg in consumer.ConsumeAsync<WebHookPayload>().WithCancellation(cancellationToken))
{
    try
    {
        if (msg.Metadata.HasValue)
        {
            maxDelivered = msg.Metadata.Value.NumDelivered;
            streamSeq = msg.Metadata.Value.Sequence.Stream;
            _logger.LogDebug($"Deliver No: {maxDelivered}");
        }
           using var httpClient = _httpClientFactory.CreateClient();
            httpClient.Timeout = TimeSpan.FromSeconds(10);
            httpClient.BaseAddress = new Uri(msg.Data.Uri);
            httpClient.DefaultRequestHeaders.Add("Accept", "application/json");

            try
            {
                var content = new StringContent(msg.Data.Payload, Encoding.UTF8, "application/json");
                var cts = new CancellationTokenSource(TimeSpan.FromSeconds(8));
                var response = await httpClient.PostAsync("", content, cancellationToken: cts.Token);
                if (response.IsSuccessStatusCode)
                {
                 Console.WriteLine("Http Sent Success");
                    await msg.AckAsync();
                }
                else
                {
                    _logger.LogDebug($"Max Delivery: {maxDelivered} Stream Seq: {streamSeq} ID: {msg.Data.ID} DateTime: {DateTimeOffset.FromUnixTimeSeconds(msg.Data.UnixDateTime).DateTime}");
                    _logger.LogDebug($"Code: 102 Http Error: {response.StatusCode} Message {response.Content.ToString()}");
                    if (maxDelivered >= configuredMaxDeliver)
                    {
                        await msg.AckAsync();
                    }
               }
            }
            catch (Exception ex)
            {
                _logger.LogDebug("Failed Error: {Message}", ex.Message);
                if (maxDelivered >= configuredMaxDeliver)
                {
                    await msg.AckAsync();
                }
            }
    }
    catch (Exception ex)
    {
        _logger.LogDebug($"Code 303 {ex.Message}");
    }
}

[mtmk]

you could have a background processing with a limit on number of task that can run in parallel. you need to also consider if the service your posting to is slow hammering it more might not be the best option. my 2 cents.

[alberk8]

My issue is with calling external HTTP, I have no control over if the site is up or slow. I also don't like the background queue with callback, logging is lost.

I tried to wrap the http call in a Task.Run() and it seems to work but if the messages are large in numbers it think it would fall over.

[mtmk]

something like polly might help too

[mtmk]

just an idea of course 😅 hopefully there will be other suggestions by others