in src/DotPulsar/Internal/ConsumerProcess.cs [42:83]
protected override void CalculateState()
{
if (_stateManager.IsFinalState())
return;
if (ExecutorState == ExecutorState.Faulted)
{
var formerState = _stateManager.SetState(ConsumerState.Faulted);
if (formerState != ConsumerState.Faulted)
ActionQueue.Enqueue(async _ => await _subConsumer.ChannelFaulted(Exception!).ConfigureAwait(false));
return;
}
switch (ChannelState)
{
case ChannelState.Active:
_stateManager.SetState(ConsumerState.Active);
return;
case ChannelState.Inactive:
_stateManager.SetState(ConsumerState.Inactive);
return;
case ChannelState.ClosedByServer:
case ChannelState.Disconnected:
_stateManager.SetState(ConsumerState.Disconnected);
ActionQueue.Enqueue(async x =>
{
await _subConsumer.CloseChannel(x).ConfigureAwait(false);
await _subConsumer.EstablishNewChannel(x).ConfigureAwait(false);
});
return;
case ChannelState.Connected:
if (!_isFailoverSubscription)
_stateManager.SetState(ConsumerState.Active);
return;
case ChannelState.ReachedEndOfTopic:
_stateManager.SetState(ConsumerState.ReachedEndOfTopic);
return;
case ChannelState.Unsubscribed:
_stateManager.SetState(ConsumerState.Unsubscribed);
return;
}
}