in src/AmqpSession.cs [854:924]
public void DisposeDelivery(Delivery delivery, bool settled, DeliveryState state, bool noFlush)
{
if (delivery.Settled)
{
this.OnWindowMoved(1);
return;
}
bool scheduleTimer = false;
Delivery toDispose = null;
lock (this.syncRoot)
{
delivery.StateChanged = true;
delivery.Settled = settled;
delivery.State = state;
if (!delivery.Batchable)
{
delivery.StateChanged = false;
toDispose = delivery;
if (delivery.Settled)
{
Delivery.Remove(ref this.firstUnsettled, ref this.lastUnsettled, delivery);
}
}
else if (this.sendingDisposition || noFlush)
{
return;
}
else if (this.session.settings.DispositionInterval == TimeSpan.Zero ||
++this.needDispositionCount >= this.session.settings.DispositionThreshold)
{
this.sendingDisposition = true;
this.needDispositionCount = 0;
}
else if (!this.timerScheduled)
{
this.timerScheduled = true;
scheduleTimer = true;
}
}
if (toDispose != null)
{
this.SendDisposition(new DispositionInfo(toDispose));
if (delivery.Settled)
{
this.OnWindowMoved(1);
}
}
else if (scheduleTimer)
{
if (this.dispositionTimer == null)
{
this.dispositionTimer = new Timer(
s => DispositionTimerCallback(s),
this,
this.session.settings.DispositionInterval,
Timeout.InfiniteTimeSpan);
}
else
{
this.dispositionTimer.Change(this.session.settings.DispositionInterval, Timeout.InfiniteTimeSpan);
}
}
else
{
this.SendDisposition();
}
}