Monitoring

Monitoring

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

Monitoring is recommended because DotPulsar will retry/reconnect on certain fault scenarios indefinitely (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.

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

Monitoring the state is easy, so let's see how to monitor when a consumer changes state:

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

    while (true)
    {
        state = await consumer.StateChangedFrom(state, cancellationToken);

        switch (state)
        {
            case ConsumerState.Active:
                Console.WriteLine("Consumer is active");
                break;
            case ConsumerState.Inactive:
                Console.WriteLine("Consumer is inactive");
                break;
            case ConsumerState.Disconnected:
                Console.WriteLine("Consumer is disconnected");
                break;
            case ConsumerState.Closed:
                Console.WriteLine("Consumer has closed");
                break;
            case ConsumerState.ReachedEndOfTopic:
                Console.WriteLine("Consumer has reached end of topic");
                break;
	        case ConsumerState.Faulted:
                Console.WriteLine("Consumer has faulted");
                break;
        }

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

Here the variable 'state' will contain the new state. You can monitor going From (StateChangedFrom) and To (StateChangedTo) a 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 e.g. are monitoring going to 'Disconnected' and the consumer is 'Closed', then your task will complete and return 'Closed'.