Monitoring

Monitoring

Consumers, producers and readers all have states that can be monitored.

Monitoring the state is recommended because DotPulsar will retry/reconnect indefinitely on certain fault scenarios (while first switching to the 'Disconnected' state). Some fault conditions are considered final and they will cause DotPulsar to throw an exception and move to a final state, after which it should be disposed.

States

Let's have a look at what states they can have.

Consumer states

• Active (All is well)
• Inactive (All is well. The subscription type is 'Failover' and you are not the active consumer)
• Closed (The consumer or PulsarClient has been disposed)
• Disconnected (The connection was lost and attempts are being made to reconnect)
• Faulted (An unrecoverable error has occurred)
• ReachedEndOfTopic (No more messages will be delivered)

Producer states

• Closed (The producer or PulsarClient has been disposed)
• Connected (All is well)
• Disconnected (The connection was lost and attempts are being made to reconnect)
• Faulted (An unrecoverable error has occurred)

Reader states

• Closed (The reader or PulsarClient has been disposed)
• Connected: (All is well)
• Disconnected (The connection was lost and attempts are being made to reconnect)
• Faulted (An unrecoverable error has occurred)
• ReachedEndOfTopic (No more messages will be delivered)

How to monitor the state using StateChanged[From/To]

Monitoring the state is easy, just call StateChangedFrom or StateChangedTo on a consumer, producer or reader. Let's see how:

var stateChanged = await consumer.StateChangedFrom(ConsumerState.Active, cancellationToken);

The variable 'stateChanged' will tell us both which consumer changed state and to which state. Some states are final, meaning the state can no longer change. For consumers 'Closed', 'Faulted' and 'ReachedEndOfTopic' are final states. When the consumer enters a final state, all monitoring tasks are completed. So if you are monitoring going to 'Disconnected' and the consumer is 'Closed', then your task will complete and return 'Closed'.

Monitoring consumer state

private static async ValueTask Monitor(IConsumer consumer, CancellationToken cancellationToken)
{
    var state = ConsumerState.Disconnected;

    while (!cancellationToken.IsCancellationRequested)
    {
        var stateChanged = await consumer.StateChangedFrom(state, cancellationToken);
        state = stateChanged.ConsumerState;

        var stateMessage = state switch
        {
            ConsumerState.Active => "The consumer is active",
            ConsumerState.Inactive => "The consumer is inactive",
            ConsumerState.Disconnected => "The consumer is disconnected",
            ConsumerState.Closed => "The consumer has closed",
            ConsumerState.ReachedEndOfTopic => "The consumer has reached end of topic",
            ConsumerState.Faulted => "The consumer has faulted",
            _ => $"The consumer has an unknown state '{state}'"
        };

        Console.WriteLine(stateMessage);

        if (consumer.IsFinalState(state))
            return;
    }
}

Monitoring reader state

private static async ValueTask Monitor(IReader reader, CancellationToken cancellationToken)
{
    var state = ReaderState.Disconnected;

    while (!cancellationToken.IsCancellationRequested)
    {
        var stateChanged = await reader.StateChangedFrom(state, cancellationToken);
        state = stateChanged.ReaderState;

        var stateMessage = state switch
        {
            ReaderState.Connected => "The reader is connected",
            ReaderState.Disconnected => "The reader is disconnected",
            ReaderState.Closed => "The reader has closed",
            ReaderState.ReachedEndOfTopic => "The reader has reached end of topic",
            ReaderState.Faulted => "The reader has faulted",
            _ => $"The reader has an unknown state '{state}'"
        };

        Console.WriteLine(stateMessage);

        if (reader.IsFinalState(state))
            return;
    }
}

Monitoring producer state

private static async ValueTask Monitor(IProducer producer, CancellationToken cancellationToken)
{
    var state = ProducerState.Disconnected;

    while (!cancellationToken.IsCancellationRequested)
    {
        var stateChanged = await producer.StateChangedFrom(state, cancellationToken);
        state = stateChanged.ProducerState;

        var stateMessage = state switch
        {
            ProducerState.Connected => $"The producer is connected",
            ProducerState.Disconnected => $"The producer is disconnected",
            ProducerState.Closed => $"The producer has closed",
            ProducerState.Faulted => $"The producer has faulted",
            _ => $"The producer has an unknown state '{state}'"
        };

        Console.WriteLine(stateMessage);

        if (producer.IsFinalState(state))
            return;
    }
}

How to monitor the state using IHandleStateChanged

Using StateChanged[From/To] can be a bit verbose, so we provide a shortcut using either IHandleStateChanged (if you want to implement your own handler) or the StateChangedHandler methods on the builders. The state is always first 'Disconnected' and the handler will be called for the first time when the state moves away from 'Disconnected'. Once the state is final the handler is called for the last time. The sample applications (Consuming, Reading, and Producing) are using this method.

await using var consumer = client.NewConsumer()
                                 .StateChangedHandler(Monitor)
                                 .SubscriptionName("MySubscription")
                                 .Topic("persistent://public/default/mytopic")
                                 .Create();

Monitoring consumer state

private static void Monitor(ConsumerStateChanged stateChanged, CancellationToken cancellationToken)
{
    var stateMessage = stateChanged.ConsumerState switch
    {
        ConsumerState.Active => "is active",
        ConsumerState.Inactive => "is inactive",
        ConsumerState.Disconnected => "is disconnected",
        ConsumerState.Closed => "has closed",
        ConsumerState.ReachedEndOfTopic => "has reached end of topic",
        ConsumerState.Faulted => "has faulted",
        _ => $"has an unknown state '{stateChanged.ConsumerState}'"
    };

    var topic = stateChanged.Consumer.Topic;
    Console.WriteLine($"The consumer for topic '{topic}' " + stateMessage);
}

Monitoring reader state

private static void Monitor(ReaderStateChanged stateChanged, CancellationToken cancellationToken)
{
    var stateMessage = stateChanged.ReaderState switch
    {
        ReaderState.Connected => "is connected",
        ReaderState.Disconnected => "is disconnected",
        ReaderState.Closed => "has closed",
        ReaderState.ReachedEndOfTopic => "has reached end of topic",
        ReaderState.Faulted => "has faulted",
        _ => $"has an unknown state '{stateChanged.ReaderState}'"
    };

    var topic = stateChanged.Reader.Topic;
    Console.WriteLine($"The reader for topic '{topic}' " + stateMessage);
}

Monitoring producer state

private static void Monitor(ProducerStateChanged stateChanged, CancellationToken cancellationToken)
{
    var stateMessage = stateChanged.ProducerState switch
    {
        ProducerState.Connected => "is connected",
        ProducerState.Disconnected => "is disconnected",
        ProducerState.Closed => "has closed",
        ProducerState.Faulted => "has faulted",
        _ => $"has an unknown state '{stateChanged.ProducerState}'"
    };

    var topic = stateChanged.Producer.Topic;
    Console.WriteLine($"The producer for topic '{topic}' " + stateMessage);
}