From 54f78408d709932c6105cb8752ac7d99d34a03a9 Mon Sep 17 00:00:00 2001 From: Luke Bakken Date: Tue, 12 May 2020 08:47:32 -0700 Subject: [PATCH] Throw ObjectDisposedException when required https://stackoverflow.com/a/50003834 Fixes #829 --- .../client/impl/AutorecoveringModel.cs | 789 +++++++++++++++++- 1 file changed, 770 insertions(+), 19 deletions(-) diff --git a/projects/RabbitMQ.Client/client/impl/AutorecoveringModel.cs b/projects/RabbitMQ.Client/client/impl/AutorecoveringModel.cs index 267f50b609..0adce230fc 100644 --- a/projects/RabbitMQ.Client/client/impl/AutorecoveringModel.cs +++ b/projects/RabbitMQ.Client/client/impl/AutorecoveringModel.cs @@ -48,6 +48,7 @@ namespace RabbitMQ.Client.Impl { internal sealed class AutorecoveringModel : IFullModel, IRecoverable { + private bool _disposed = false; private readonly object _eventLock = new object(); private AutorecoveringConnection _connection; private RecoveryAwareModel _delegate; @@ -65,13 +66,38 @@ internal sealed class AutorecoveringModel : IFullModel, IRecoverable public IConsumerDispatcher ConsumerDispatcher { - get { return _delegate.ConsumerDispatcher; } + get + { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + + return _delegate.ConsumerDispatcher; + } } public TimeSpan ContinuationTimeout { - get { return _delegate.ContinuationTimeout; } - set { _delegate.ContinuationTimeout = value; } + get + { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + + return _delegate.ContinuationTimeout; + } + + set + { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + + _delegate.ContinuationTimeout = value; + } } public AutorecoveringModel(AutorecoveringConnection conn, RecoveryAwareModel _delegate) @@ -84,6 +110,11 @@ public event EventHandler BasicAcks { add { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + lock (_eventLock) { _recordedBasicAckEventHandlers += value; @@ -92,6 +123,11 @@ public event EventHandler BasicAcks } remove { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + lock (_eventLock) { _recordedBasicAckEventHandlers -= value; @@ -104,6 +140,11 @@ public event EventHandler BasicNacks { add { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + lock (_eventLock) { _recordedBasicNackEventHandlers += value; @@ -112,6 +153,11 @@ public event EventHandler BasicNacks } remove { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + lock (_eventLock) { _recordedBasicNackEventHandlers -= value; @@ -124,16 +170,34 @@ public event EventHandler BasicRecoverOk { add { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + // TODO: record and re-add handlers _delegate.BasicRecoverOk += value; } - remove { _delegate.BasicRecoverOk -= value; } + remove + { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + + _delegate.BasicRecoverOk -= value; + } } public event EventHandler BasicReturn { add { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + lock (_eventLock) { _recordedBasicReturnEventHandlers += value; @@ -142,6 +206,11 @@ public event EventHandler BasicReturn } remove { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + lock (_eventLock) { _recordedBasicReturnEventHandlers -= value; @@ -154,6 +223,11 @@ public event EventHandler CallbackException { add { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + lock (_eventLock) { _recordedCallbackExceptionEventHandlers += value; @@ -162,6 +236,11 @@ public event EventHandler CallbackException } remove { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + lock (_eventLock) { _recordedCallbackExceptionEventHandlers -= value; @@ -174,16 +253,34 @@ public event EventHandler FlowControl { add { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + // TODO: record and re-add handlers _delegate.FlowControl += value; } - remove { _delegate.FlowControl -= value; } + remove + { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + + _delegate.FlowControl -= value; + } } public event EventHandler ModelShutdown { add { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + lock (_eventLock) { _recordedShutdownEventHandlers += value; @@ -192,6 +289,11 @@ public event EventHandler ModelShutdown } remove { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + lock (_eventLock) { _recordedShutdownEventHandlers -= value; @@ -204,42 +306,115 @@ public event EventHandler ModelShutdown public int ChannelNumber { - get { return _delegate.ChannelNumber; } + get + { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + + return _delegate.ChannelNumber; + } } public ShutdownEventArgs CloseReason { - get { return _delegate.CloseReason; } + get + { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + + return _delegate.CloseReason; + } } public IBasicConsumer DefaultConsumer { - get { return _delegate.DefaultConsumer; } - set { _delegate.DefaultConsumer = value; } + get + { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + + return _delegate.DefaultConsumer; + } + set + { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + + _delegate.DefaultConsumer = value; + } } public IModel Delegate { - get { return _delegate; } + get + { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + + return _delegate; + } } public bool IsClosed { - get { return _delegate.IsClosed; } + get + { + if (_delegate == null) + { + return false; + } + else + { + return _delegate.IsClosed; + } + } } public bool IsOpen { - get { return _delegate.IsOpen; } + get + { + if (_delegate == null) + { + return false; + } + else + { + return _delegate.IsOpen; + } + } } public ulong NextPublishSeqNo { - get { return _delegate.NextPublishSeqNo; } + get + { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + + return _delegate.NextPublishSeqNo; + } } public void AutomaticallyRecover(AutorecoveringConnection conn) { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + _connection = conn; RecoveryAwareModel defunctModel = _delegate; @@ -260,11 +435,21 @@ public void AutomaticallyRecover(AutorecoveringConnection conn) public void BasicQos(ushort prefetchCount, bool global) { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + _delegate.BasicQos(0, prefetchCount, global); } public void Close(ushort replyCode, string replyText, bool abort) { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + try { _delegate.Close(replyCode, replyText, abort); @@ -277,6 +462,11 @@ public void Close(ushort replyCode, string replyText, bool abort) public void Close(ShutdownEventArgs reason, bool abort) { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + try { _delegate.Close(reason, abort).GetAwaiter().GetResult();; @@ -289,66 +479,131 @@ public void Close(ShutdownEventArgs reason, bool abort) public bool DispatchAsynchronous(Command cmd) { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + return _delegate.DispatchAsynchronous(cmd); } public void FinishClose() { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + _delegate.FinishClose(); } public void HandleCommand(ISession session, Command cmd) { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + _delegate.HandleCommand(session, cmd); } public void OnBasicAck(BasicAckEventArgs args) { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + _delegate.OnBasicAck(args); } public void OnBasicNack(BasicNackEventArgs args) { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + _delegate.OnBasicNack(args); } public void OnBasicRecoverOk(EventArgs args) { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + _delegate.OnBasicRecoverOk(args); } public void OnBasicReturn(BasicReturnEventArgs args) { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + _delegate.OnBasicReturn(args); } public void OnCallbackException(CallbackExceptionEventArgs args) { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + _delegate.OnCallbackException(args); } public void OnFlowControl(FlowControlEventArgs args) { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + _delegate.OnFlowControl(args); } public void OnModelShutdown(ShutdownEventArgs reason) { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + _delegate.OnModelShutdown(reason); } public void OnSessionShutdown(ISession session, ShutdownEventArgs reason) { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + _delegate.OnSessionShutdown(session, reason); } public bool SetCloseReason(ShutdownEventArgs reason) { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + return _delegate.SetCloseReason(reason); } public override string ToString() { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + return _delegate.ToString(); } @@ -359,9 +614,13 @@ void IDisposable.Dispose() private void Dispose(bool disposing) { + if (_disposed) + { + return; + } + if (disposing) { - // dispose managed resources Abort(); _connection = null; @@ -371,36 +630,61 @@ private void Dispose(bool disposing) _recordedBasicReturnEventHandlers = null; _recordedCallbackExceptionEventHandlers = null; _recordedShutdownEventHandlers = null; - } - // dispose unmanaged resources + _disposed = true; + } } public void ConnectionTuneOk(ushort channelMax, uint frameMax, ushort heartbeat) { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + _delegate.ConnectionTuneOk(channelMax, frameMax, heartbeat); } public void HandleBasicAck(ulong deliveryTag, bool multiple) { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + _delegate.HandleBasicAck(deliveryTag, multiple); } public void HandleBasicCancel(string consumerTag, bool nowait) { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + _delegate.HandleBasicCancel(consumerTag, nowait); } public void HandleBasicCancelOk(string consumerTag) { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + _delegate.HandleBasicCancelOk(consumerTag); } public void HandleBasicConsumeOk(string consumerTag) { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + _delegate.HandleBasicConsumeOk(consumerTag); } @@ -412,11 +696,24 @@ public void HandleBasicDeliver(string consumerTag, IBasicProperties basicProperties, ReadOnlyMemory body) { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + _delegate.HandleBasicDeliver(consumerTag, deliveryTag, redelivered, exchange, routingKey, basicProperties, body); } - public void HandleBasicGetEmpty() => _delegate.HandleBasicGetEmpty(); + public void HandleBasicGetEmpty() + { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + + _delegate.HandleBasicGetEmpty(); + } public void HandleBasicGetOk(ulong deliveryTag, bool redelivered, @@ -426,6 +723,11 @@ public void HandleBasicGetOk(ulong deliveryTag, IBasicProperties basicProperties, ReadOnlyMemory body) { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + _delegate.HandleBasicGetOk(deliveryTag, redelivered, exchange, routingKey, messageCount, basicProperties, body); } @@ -434,11 +736,21 @@ public void HandleBasicNack(ulong deliveryTag, bool multiple, bool requeue) { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + _delegate.HandleBasicNack(deliveryTag, multiple, requeue); } public void HandleBasicRecoverOk() { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + _delegate.HandleBasicRecoverOk(); } @@ -449,6 +761,11 @@ public void HandleBasicReturn(ushort replyCode, IBasicProperties basicProperties, ReadOnlyMemory body) { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + _delegate.HandleBasicReturn(replyCode, replyText, exchange, routingKey, basicProperties, body); } @@ -458,21 +775,41 @@ public void HandleChannelClose(ushort replyCode, ushort classId, ushort methodId) { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + _delegate.HandleChannelClose(replyCode, replyText, classId, methodId); } public void HandleChannelCloseOk() { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + _delegate.HandleChannelCloseOk(); } public void HandleChannelFlow(bool active) { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + _delegate.HandleChannelFlow(active); } public void HandleConnectionBlocked(string reason) { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + _delegate.HandleConnectionBlocked(reason); } @@ -481,16 +818,31 @@ public void HandleConnectionClose(ushort replyCode, ushort classId, ushort methodId) { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + _delegate.HandleConnectionClose(replyCode, replyText, classId, methodId); } public void HandleConnectionOpenOk(string knownHosts) { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + _delegate.HandleConnectionOpenOk(knownHosts); } public void HandleConnectionSecure(byte[] challenge) { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + _delegate.HandleConnectionSecure(challenge); } @@ -500,6 +852,11 @@ public void HandleConnectionStart(byte versionMajor, byte[] mechanisms, byte[] locales) { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + _delegate.HandleConnectionStart(versionMajor, versionMinor, serverProperties, mechanisms, locales); } @@ -508,11 +865,21 @@ public void HandleConnectionTune(ushort channelMax, uint frameMax, ushort heartbeat) { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + _delegate.HandleConnectionTune(channelMax, frameMax, heartbeat); } public void HandleConnectionUnblocked() { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + _delegate.HandleConnectionUnblocked(); } @@ -520,14 +887,23 @@ public void HandleQueueDeclareOk(string queue, uint messageCount, uint consumerCount) { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + _delegate.HandleQueueDeclareOk(queue, messageCount, consumerCount); } public void _Private_BasicCancel(string consumerTag, bool nowait) { - _delegate._Private_BasicCancel(consumerTag, - nowait); + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + + _delegate._Private_BasicCancel(consumerTag, nowait); } public void _Private_BasicConsume(string queue, @@ -538,6 +914,11 @@ public void _Private_BasicConsume(string queue, bool nowait, IDictionary arguments) { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + _delegate._Private_BasicConsume(queue, consumerTag, noLocal, @@ -549,6 +930,11 @@ public void _Private_BasicConsume(string queue, public void _Private_BasicGet(string queue, bool autoAck) { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + _delegate._Private_BasicGet(queue, autoAck); } @@ -563,12 +949,22 @@ public void _Private_BasicPublish(string exchange, throw new ArgumentNullException(nameof(routingKey)); } + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + _delegate._Private_BasicPublish(exchange, routingKey, mandatory, basicProperties, body); } public void _Private_BasicRecover(bool requeue) { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + _delegate._Private_BasicRecover(requeue); } @@ -577,27 +973,52 @@ public void _Private_ChannelClose(ushort replyCode, ushort classId, ushort methodId) { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + _delegate._Private_ChannelClose(replyCode, replyText, classId, methodId); } public void _Private_ChannelCloseOk() { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + _delegate._Private_ChannelCloseOk(); } public void _Private_ChannelFlowOk(bool active) { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + _delegate._Private_ChannelFlowOk(active); } public void _Private_ChannelOpen(string outOfBand) { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + _delegate._Private_ChannelOpen(outOfBand); } public void _Private_ConfirmSelect(bool nowait) { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + _delegate._Private_ConfirmSelect(nowait); } @@ -606,12 +1027,22 @@ public void _Private_ConnectionClose(ushort replyCode, ushort classId, ushort methodId) { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + _delegate._Private_ConnectionClose(replyCode, replyText, classId, methodId); } public void _Private_ConnectionCloseOk() { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + _delegate._Private_ConnectionCloseOk(); } @@ -619,23 +1050,43 @@ public void _Private_ConnectionOpen(string virtualHost, string capabilities, bool insist) { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + _delegate._Private_ConnectionOpen(virtualHost, capabilities, insist); } public void _Private_ConnectionSecureOk(byte[] response) { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + _delegate._Private_ConnectionSecureOk(response); } public void _Private_ConnectionStartOk(IDictionary clientProperties, string mechanism, byte[] response, string locale) { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + _delegate._Private_ConnectionStartOk(clientProperties, mechanism, response, locale); } public void _Private_UpdateSecret(byte[] newSecret, string reason) { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + _delegate._Private_UpdateSecret(newSecret, reason); } @@ -645,6 +1096,11 @@ public void _Private_ExchangeBind(string destination, bool nowait, IDictionary arguments) { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + _delegate._Private_ExchangeBind(destination, source, routingKey, nowait, arguments); } @@ -658,6 +1114,11 @@ public void _Private_ExchangeDeclare(string exchange, bool nowait, IDictionary arguments) { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + _delegate._Private_ExchangeDeclare(exchange, type, passive, durable, autoDelete, @internal, nowait, arguments); @@ -667,6 +1128,11 @@ public void _Private_ExchangeDelete(string exchange, bool ifUnused, bool nowait) { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + _delegate._Private_ExchangeDelete(exchange, ifUnused, nowait); } @@ -676,6 +1142,11 @@ public void _Private_ExchangeUnbind(string destination, bool nowait, IDictionary arguments) { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + _delegate._Private_ExchangeUnbind(destination, source, routingKey, nowait, arguments); } @@ -686,6 +1157,11 @@ public void _Private_QueueBind(string queue, bool nowait, IDictionary arguments) { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + _delegate._Private_QueueBind(queue, exchange, routingKey, nowait, arguments); } @@ -698,6 +1174,11 @@ public void _Private_QueueDeclare(string queue, bool nowait, IDictionary arguments) { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + _delegate._Private_QueueDeclare(queue, passive, durable, exclusive, autoDelete, nowait, arguments); @@ -708,6 +1189,11 @@ public uint _Private_QueueDelete(string queue, bool ifEmpty, bool nowait) { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + return _delegate._Private_QueueDelete(queue, ifUnused, ifEmpty, nowait); } @@ -715,11 +1201,21 @@ public uint _Private_QueueDelete(string queue, public uint _Private_QueuePurge(string queue, bool nowait) { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + return _delegate._Private_QueuePurge(queue, nowait); } public void Abort() { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + try { _delegate.Abort(); @@ -732,6 +1228,11 @@ public void Abort() public void Abort(ushort replyCode, string replyText) { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + try { _delegate.Abort(replyCode, replyText); @@ -745,11 +1246,21 @@ public void Abort(ushort replyCode, string replyText) public void BasicAck(ulong deliveryTag, bool multiple) { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + _delegate.BasicAck(deliveryTag, multiple); } public void BasicCancel(string consumerTag) { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + RecordedConsumer cons = _connection.DeleteRecordedConsumer(consumerTag); if (cons != null) { @@ -760,6 +1271,11 @@ public void BasicCancel(string consumerTag) public void BasicCancelNoWait(string consumerTag) { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + RecordedConsumer cons = _connection.DeleteRecordedConsumer(consumerTag); if (cons != null) { @@ -777,6 +1293,11 @@ public string BasicConsume( IDictionary arguments, IBasicConsumer consumer) { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + string result = _delegate.BasicConsume(queue, autoAck, consumerTag, noLocal, exclusive, arguments, consumer); RecordedConsumer rc = new RecordedConsumer(this, queue). @@ -792,6 +1313,11 @@ public string BasicConsume( public BasicGetResult BasicGet(string queue, bool autoAck) { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + return _delegate.BasicGet(queue, autoAck); } @@ -799,6 +1325,11 @@ public void BasicNack(ulong deliveryTag, bool multiple, bool requeue) { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + _delegate.BasicNack(deliveryTag, multiple, requeue); } @@ -813,6 +1344,11 @@ public void BasicPublish(string exchange, throw new ArgumentNullException(nameof(routingKey)); } + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + _delegate.BasicPublish(exchange, routingKey, mandatory, @@ -824,6 +1360,11 @@ public void BasicQos(uint prefetchSize, ushort prefetchCount, bool global) { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + if (global) { _prefetchCountGlobal = prefetchCount; @@ -837,22 +1378,42 @@ public void BasicQos(uint prefetchSize, public void BasicRecover(bool requeue) { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + _delegate.BasicRecover(requeue); } public void BasicRecoverAsync(bool requeue) { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + _delegate.BasicRecoverAsync(requeue); } public void BasicReject(ulong deliveryTag, bool requeue) { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + _delegate.BasicReject(deliveryTag, requeue); } public void Close() { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + try { _delegate.Close(); @@ -865,6 +1426,11 @@ public void Close() public void Close(ushort replyCode, string replyText) { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + try { _delegate.Close(replyCode, replyText); @@ -877,12 +1443,22 @@ public void Close(ushort replyCode, string replyText) public void ConfirmSelect() { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + _usesPublisherConfirms = true; _delegate.ConfirmSelect(); } public IBasicProperties CreateBasicProperties() { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + return _delegate.CreateBasicProperties(); } @@ -891,6 +1467,11 @@ public void ExchangeBind(string destination, string routingKey, IDictionary arguments) { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + RecordedBinding eb = new RecordedExchangeBinding(this). WithSource(source). WithDestination(destination). @@ -905,12 +1486,22 @@ public void ExchangeBindNoWait(string destination, string routingKey, IDictionary arguments) { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + _delegate.ExchangeBindNoWait(destination, source, routingKey, arguments); } public void ExchangeDeclare(string exchange, string type, bool durable, bool autoDelete, IDictionary arguments) { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + RecordedExchange rx = new RecordedExchange(this, exchange). WithType(type). WithDurable(durable). @@ -927,6 +1518,11 @@ public void ExchangeDeclareNoWait(string exchange, bool autoDelete, IDictionary arguments) { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + RecordedExchange rx = new RecordedExchange(this, exchange). WithType(type). WithDurable(durable). @@ -939,12 +1535,22 @@ public void ExchangeDeclareNoWait(string exchange, public void ExchangeDeclarePassive(string exchange) { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + _delegate.ExchangeDeclarePassive(exchange); } public void ExchangeDelete(string exchange, bool ifUnused) { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + _delegate.ExchangeDelete(exchange, ifUnused); _connection.DeleteRecordedExchange(exchange); } @@ -952,6 +1558,11 @@ public void ExchangeDelete(string exchange, public void ExchangeDeleteNoWait(string exchange, bool ifUnused) { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + _delegate.ExchangeDeleteNoWait(exchange, ifUnused); _connection.DeleteRecordedExchange(exchange); } @@ -961,6 +1572,11 @@ public void ExchangeUnbind(string destination, string routingKey, IDictionary arguments) { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + RecordedBinding eb = new RecordedExchangeBinding(this). WithSource(source). WithDestination(destination). @@ -976,6 +1592,11 @@ public void ExchangeUnbindNoWait(string destination, string routingKey, IDictionary arguments) { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + _delegate.ExchangeUnbind(destination, source, routingKey, arguments); } @@ -984,6 +1605,11 @@ public void QueueBind(string queue, string routingKey, IDictionary arguments) { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + RecordedBinding qb = new RecordedQueueBinding(this). WithSource(exchange). WithDestination(queue). @@ -998,6 +1624,11 @@ public void QueueBindNoWait(string queue, string routingKey, IDictionary arguments) { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + _delegate.QueueBind(queue, exchange, routingKey, arguments); } @@ -1005,6 +1636,11 @@ public QueueDeclareOk QueueDeclare(string queue, bool durable, bool exclusive, bool autoDelete, IDictionary arguments) { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + QueueDeclareOk result = _delegate.QueueDeclare(queue, durable, exclusive, autoDelete, arguments); RecordedQueue rq = new RecordedQueue(this, result.QueueName). @@ -1021,6 +1657,11 @@ public void QueueDeclareNoWait(string queue, bool durable, bool exclusive, bool autoDelete, IDictionary arguments) { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + _delegate.QueueDeclareNoWait(queue, durable, exclusive, autoDelete, arguments); RecordedQueue rq = new RecordedQueue(this, queue). @@ -1034,16 +1675,31 @@ public void QueueDeclareNoWait(string queue, bool durable, public QueueDeclareOk QueueDeclarePassive(string queue) { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + return _delegate.QueueDeclarePassive(queue); } public uint MessageCount(string queue) { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + return _delegate.MessageCount(queue); } public uint ConsumerCount(string queue) { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + return _delegate.ConsumerCount(queue); } @@ -1051,6 +1707,11 @@ public uint QueueDelete(string queue, bool ifUnused, bool ifEmpty) { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + uint result = _delegate.QueueDelete(queue, ifUnused, ifEmpty); _connection.DeleteRecordedQueue(queue); return result; @@ -1060,12 +1721,22 @@ public void QueueDeleteNoWait(string queue, bool ifUnused, bool ifEmpty) { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + _delegate.QueueDeleteNoWait(queue, ifUnused, ifEmpty); _connection.DeleteRecordedQueue(queue); } public uint QueuePurge(string queue) { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + return _delegate.QueuePurge(queue); } @@ -1074,6 +1745,11 @@ public void QueueUnbind(string queue, string routingKey, IDictionary arguments) { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + RecordedBinding qb = new RecordedQueueBinding(this). WithSource(exchange). WithDestination(queue). @@ -1086,47 +1762,92 @@ public void QueueUnbind(string queue, public void TxCommit() { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + _delegate.TxCommit(); } public void TxRollback() { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + _delegate.TxRollback(); } public void TxSelect() { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + _usesTransactions = true; _delegate.TxSelect(); } public bool WaitForConfirms(TimeSpan timeout, out bool timedOut) { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + return _delegate.WaitForConfirms(timeout, out timedOut); } public bool WaitForConfirms(TimeSpan timeout) { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + return _delegate.WaitForConfirms(timeout); } public bool WaitForConfirms() { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + return _delegate.WaitForConfirms(); } public void WaitForConfirmsOrDie() { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + _delegate.WaitForConfirmsOrDie(); } public void WaitForConfirmsOrDie(TimeSpan timeout) { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + _delegate.WaitForConfirmsOrDie(timeout); } private void RecoverBasicAckHandlers() { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + lock (_eventLock) { _delegate.BasicAcks += _recordedBasicAckEventHandlers; @@ -1135,6 +1856,11 @@ private void RecoverBasicAckHandlers() private void RecoverBasicNackHandlers() { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + lock (_eventLock) { _delegate.BasicNacks += _recordedBasicNackEventHandlers; @@ -1143,6 +1869,11 @@ private void RecoverBasicNackHandlers() private void RecoverBasicReturnHandlers() { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + lock (_eventLock) { _delegate.BasicReturn += _recordedBasicReturnEventHandlers; @@ -1151,6 +1882,11 @@ private void RecoverBasicReturnHandlers() private void RecoverCallbackExceptionHandlers() { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + lock (_eventLock) { _delegate.CallbackException += _recordedCallbackExceptionEventHandlers; @@ -1159,6 +1895,11 @@ private void RecoverCallbackExceptionHandlers() private void RecoverModelShutdownHandlers() { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + lock (_eventLock) { _delegate.ModelShutdown += _recordedShutdownEventHandlers; @@ -1190,6 +1931,11 @@ private void RecoverState() private void RunRecoveryEventHandlers() { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + foreach (EventHandler reh in Recovery?.GetInvocationList() ?? Array.Empty()) { try @@ -1207,6 +1953,11 @@ private void RunRecoveryEventHandlers() public IBasicPublishBatch CreateBasicPublishBatch() { + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } + return ((IFullModel)_delegate).CreateBasicPublishBatch(); } }