C# consumer flow dispose

Hey Ragnar

Thank you for your response.

With a window size of 255, it seems to take anywhere between 0 and 60s to dispose.

With a window size of 1, it disposes instantly.

So handling a message takes between 1-3s, so I guess you could call it a slow-ish consumer.

Same code below + message handler code sending the signal to unblock:

_flow = session.CreateFlow(flowProperties, queue, null, HandleMessageEvent, HandleFlowEvent);
_logger.LogInformation("Subscribed to queue {MessageQueue}", Q);

do
{
    _waitEventWaitHandle.WaitOne();
} while (!stoppingToken.IsCancellationRequested && !_isSessionDown);

_flow.Stop();

var sw = Stopwatch.StartNew();

_flow.Dispose();

_logger.LogInformation("Flow disposed in {Milliseconds}ms.", sw.ElapsedMilliseconds);

The message handler below:

private void HandleMessageEvent(object? source, MessageEventArgs args)
{
    try
    {
        _logger.LogInformation("Received message.");

        using var message = args.Message;

        // Process message. This would average between 1-3 seconds.

        _flow.Ack(message.ADMessageId);


        _logger.LogInformation("Message acknowledged.");
    } catch (Exception ex)
    {
        _logger.LogError("Exception thrown: {0}", ex.Message);
    }
    finally
    {
        _waitEventWaitHandle.Set();
    }
}

I do see messages still being handled while the Dispose is 'blocked'. Its like its trying to drain the flow before disposing.

Hope this helps.

Thanks

Shaun

Hi Ragnar

Thanks again for the response.

Just to be clear. I want to process more than one message per flow, this is why I have the do...while loop around the wait/blocking operation. See below:

do
{
    _waitEventWaitHandle.WaitOne();
} while (!stoppingToken.IsCancellationRequested && !_isSessionDown);

Every time a message is handled, I send a signal to the blocked thread, In order to evaluate the cancellation token - just in case the application is trying to shut down gracefully.

I might need to go back to the drawing board. The reason I am doing the 'slow' processing in the callback, is because I want to make sure that I do process the message before ack-ing it. Perhaps I should rather persist these message 'instructions' to store (db or persisted cache), then process them in another process. This way if the application dies, I can recover those 'instructions' from store. Also persisting these messages and not processing them, should put less strain on the callback.

Would you agree?

Thanks

Shaun

Thanks Ragnar,

That does make sense.

I am still a bit unclear on how the use the Flow concept optimally.

Is it acceptable to have one Flow instance for the duration of the application?

The only reason I would want to stop/dispose the flow & re-create, is if I have a message ack that failed. My understanding is that creating a new flow will send those un-acked message again.

Thanks

Shaun