Better Connect and let incorrect logging

GrumpyBusted · GrumpyBusted · commit 618d08f755c4 · 2018-02-02T21:49:24.000+01:00
diff --git a/Grumpy.MessageQueue.Msmq.IntegrationTests/LocaleQueueTests.cs b/Grumpy.MessageQueue.Msmq.IntegrationTests/LocaleQueueTests.cs
@@ -146,7 +146,7 @@ public void CancelReceiveAsyncShouldStopBeforeTimeout()
                     }
 
                     stopwatch.Stop();
-                    stopwatch.ElapsedMilliseconds.Should().BeInRange(400, 700);
+                    stopwatch.ElapsedMilliseconds.Should().BeInRange(400, 711);
                 }
             }
             finally
@@ -270,6 +270,7 @@ public void SendAndReceiveLargeMessageShouldWork()
 
                     queue.Count().Should().Be(2);
                     queue.Transactional.Should().BeTrue();
+                    queue.AccessMode.Should().Be(AccessMode.SendAndReceive);
                 }
 
                 using (var queue = CreateLocalQueue(name, true, LocaleQueueMode.Durable))
diff --git a/Grumpy.MessageQueue.Msmq.UnitTests/QueueTests.cs b/Grumpy.MessageQueue.Msmq.UnitTests/QueueTests.cs
@@ -50,6 +50,17 @@ public void SendToExistingQueueShouldNotCallCreate()
             _messageQueueManager.Received(2).Get(".", Arg.Any<string>(), Arg.Any<bool>(), Arg.Any<QueueAccessMode>());
         }
 
+        [Fact]
+        public void SendLargeMessageToNonTransactionalQueueShouldThrow()
+        {
+            SetQueue(Substitute.For<System.Messaging.MessageQueue>(), true);
+
+            using (var cut = CreateLocaleQueue("MyQueue", true, LocaleQueueMode.Durable, AccessMode.Send, false))
+            {
+                Assert.Throws<MessageSizeException>(() => cut.Send(new StringBuilder().Insert(0, "A", 5000000)));
+            }
+        }
+
         [Fact]
         public void SendToNoneExistingQueueShouldCallCreate()
         {
@@ -273,9 +284,9 @@ public void ReceiveOnSendQueueShouldThrowException()
             }
         }
 
-        private IQueue CreateLocaleQueue(string queue = "MyQueue", bool privateQueue = true, LocaleQueueMode localeQueueMode = LocaleQueueMode.TemporaryMaster, AccessMode accessMode = AccessMode.SendAndReceive)
+        private IQueue CreateLocaleQueue(string queue = "MyQueue", bool privateQueue = true, LocaleQueueMode localeQueueMode = LocaleQueueMode.TemporaryMaster, AccessMode accessMode = AccessMode.SendAndReceive, bool transactional = true)
         {
-            return new LocaleQueue(_logger, _messageQueueManager, _messageQueueTransactionFactory, queue, privateQueue, localeQueueMode, true, accessMode);
+            return new LocaleQueue(_logger, _messageQueueManager, _messageQueueTransactionFactory, queue, privateQueue, localeQueueMode, transactional, accessMode);
         }
 
         private void SetQueue(System.Messaging.MessageQueue queue, bool exists)
diff --git a/Grumpy.MessageQueue.Msmq/Queue.cs b/Grumpy.MessageQueue.Msmq/Queue.cs
@@ -16,7 +16,6 @@
 using Grumpy.MessageQueue.Msmq.Interfaces;
 using Microsoft.Extensions.Logging;
 using Newtonsoft.Json;
-using ObjectExtensions = Grumpy.Common.Extensions.ObjectExtensions;
 
 namespace Grumpy.MessageQueue.Msmq
 {
@@ -25,7 +24,6 @@ public abstract class Queue : IQueue
     {
         private const int MaxMsmqMessageSize = 4096000;
         private System.Messaging.MessageQueue _messageQueue;
-        private readonly AccessMode _accessMode;
         private readonly object _messageQueueLock;
         private readonly Timer _disconnectTimer;
 
@@ -49,7 +47,7 @@ protected Queue(ILogger logger, IMessageQueueManager messageQueueManager, IMessa
             Durable = durable;
             _messageQueueLock = new object();
             _disconnectTimer = new Timer(Disconnect, null, 3600000, 3600000);
-            _accessMode = accessMode;
+            AccessMode = accessMode;
         }
 
         /// <summary>
@@ -80,11 +78,14 @@ protected Queue(ILogger logger, IMessageQueueManager messageQueueManager, IMessa
         /// <inheritdoc />
         public bool Transactional { get; }
 
+        /// <inheritdoc />
+        public AccessMode AccessMode { get; }
+
         /// <inheritdoc />
         public int Count()
         {
-            if (!_accessMode.In(AccessMode.Receive, AccessMode.SendAndReceive))
-                throw new AccessModeException(nameof(Count), _accessMode);
+            if (!AccessMode.In(AccessMode.Receive, AccessMode.SendAndReceive))
+                throw new AccessModeException(nameof(Count), AccessMode);
 
             try
             {
@@ -123,16 +124,13 @@ public virtual void Connect()
         {
             lock (_messageQueueLock)
             {
-                if (_messageQueue != null)
-                    Disconnect();
-
                 if (_messageQueue == null)
                 {
-                    _messageQueue = GetQueue(_accessMode);
+                    _messageQueue = GetQueue(AccessMode);
 
                     if (_messageQueue != null)
                     {
-                        switch (_accessMode)
+                        switch (AccessMode)
                         {
                             case AccessMode.Receive:
                                 _messageQueue.MessageReadPropertyFilter = new MessagePropertyFilter { AppSpecific = true, Id = true, Body = true };
@@ -147,11 +145,11 @@ public virtual void Connect()
                                 _messageQueue.DefaultPropertiesToSend = new DefaultPropertiesToSend { Recoverable = Durable };
                                 break;
                             default:
-                                throw new ArgumentOutOfRangeException(nameof(_accessMode), _accessMode, "Unknown Access Mode");
+                                throw new ArgumentOutOfRangeException(nameof(AccessMode), AccessMode, "Unknown Access Mode");
                         }
                     }
 
-                    Logger.Information("Connected to Message Queue {@Queue} {@MessageQueue}", this, _messageQueue);
+                    Logger.Information("Connected to Message Queue {@Queue}", this);
                 }
 
                 if (_messageQueue == null)
@@ -187,8 +185,8 @@ public void Disconnect()
         /// <inheritdoc />
         public void Send<T>(T message)
         {
-            if (!_accessMode.In(AccessMode.Send, AccessMode.SendAndReceive))
-                throw new AccessModeException(nameof(Send), _accessMode);
+            if (!AccessMode.In(AccessMode.Send, AccessMode.SendAndReceive))
+                throw new AccessModeException(nameof(Send), AccessMode);
 
             try
             {
@@ -248,8 +246,8 @@ public async Task<ITransactionalMessage> ReceiveAsync(int millisecondsTimeout, C
 
         private async Task<ITransactionalMessage> ReceiveAsyncInternal(int millisecondsTimeout, CancellationToken cancellationToken)
         {
-            if (!_accessMode.In(AccessMode.Receive, AccessMode.SendAndReceive))
-                throw new AccessModeException(nameof(Send), _accessMode);
+            if (!AccessMode.In(AccessMode.Receive, AccessMode.SendAndReceive))
+                throw new AccessModeException(nameof(Send), AccessMode);
 
             var timeout = TimeSpan.FromMilliseconds(millisecondsTimeout);
 
diff --git a/Grumpy.MessageQueue.TestTools.UnitTests/TestQueueHandlerTests.cs b/Grumpy.MessageQueue.TestTools.UnitTests/TestQueueHandlerTests.cs
@@ -30,6 +30,18 @@ public void TestQueueHandlerShouldWork()
             numberOfHeartbeats.Should().BeGreaterOrEqualTo(1);
         }
 
+        [Fact]
+        public void TestQueueHandlerShouldWorkWithCancelHandler()
+        {
+            var q = new TestQueueHandlerFactory();
+            using (var w = q.Create())
+            {
+                w.Start("MyQueue", true, LocaleQueueMode.DurableCreate, true, Handler, (o, e) => true, null, 1, true, true, new CancellationToken());
+                w.Stop();
+                w.Idle.Should().BeFalse();
+            }
+        }
+
         private void Handler(object message, CancellationToken cancellationToken)
         {
             if ((string)message == "Exception")
diff --git a/Grumpy.MessageQueue.UnitTests/QueueHandlerAsyncTests.cs b/Grumpy.MessageQueue.UnitTests/QueueHandlerAsyncTests.cs
@@ -46,6 +46,25 @@ public void CanStopQueue()
             }
         }
 
+        [Fact]
+        public void InvalidHeartbeatRateShouldThrow()
+        {
+            using (var cut = CreateQueueHandler())
+            {
+                Assert.Throws<ArgumentException>(() => cut.Start("MyQueue", true, LocaleQueueMode.TemporaryMaster, true, (m,c) => { }, null, () => { }, -1, false, false, _cancellationToken));
+            }
+        }
+
+        [Fact]
+        public void MultiStartShouldThrowException()
+        {
+            using (var cut = CreateQueueHandler())
+            {
+                cut.Start("MyQueue", true, LocaleQueueMode.TemporaryMaster, true, (m,c) => { }, null, null, 100, false, false, _cancellationToken);
+                Assert.Throws<ArgumentException>(() => cut.Start("MyQueue", true, LocaleQueueMode.TemporaryMaster, true, (m,c) => { }, null, null, 100, false, false, _cancellationToken));
+            }
+        }
+
         [Fact]
         public void QueueBeforeShouldBeIdle()
         {
@@ -81,6 +100,18 @@ private static ITransactionalMessage CreateMessage(object body)
             return message;
         }
 
+        [Fact]
+        public void HeartbeatAreCalled()
+        {
+            using (var cut = CreateQueueHandler())
+            {
+                var i = 0;
+                cut.Start("MyQueue", true, LocaleQueueMode.TemporaryMaster, true, (m,c) => { }, null, () => { ++i; }, 1, false, false, _cancellationToken);
+                Thread.Sleep(1000);
+                i.Should().BeGreaterOrEqualTo(1);
+            }
+        }
+
         public void Dispose()
         {
             Dispose(true);
diff --git a/Grumpy.MessageQueue.UnitTests/QueueHandlerSyncTests.cs b/Grumpy.MessageQueue.UnitTests/QueueHandlerSyncTests.cs
@@ -56,6 +56,19 @@ public void HandlerThrowExceptionShouldResultInErrors()
             numberOfErrors.Should().Be(1);
         }
 
+        [Fact]
+        public void HandlerAndErrorHandlerThrowExceptionShouldNAckMessage()
+        {
+            var numberOfErrors = 0;
+            var message = CreateMessage("MyMessage");
+
+            _queue.Receive(Arg.Any<int>(), Arg.Any<CancellationToken>()).Returns(e => message, e => null);
+
+            ExecuteHandler((m, c) => throw new Exception((string)m), (m, e) => throw new Exception());
+
+            message.Received(1).NAck();
+        }
+
         [Fact]
         public void HeartbeatHandlerShouldBeCalled()
         {
diff --git a/Grumpy.MessageQueue/Interfaces/IQueue.cs b/Grumpy.MessageQueue/Interfaces/IQueue.cs
@@ -31,6 +31,11 @@ public interface IQueue : IDisposable
         /// </summary>
         bool Transactional { get; }
 
+        /// <summary>
+        /// Queue Access Mode
+        /// </summary>
+        AccessMode AccessMode { get; }
+
         /// <summary>
         /// Number of Messages in Queue
         /// </summary>

Original file line number	Diff line number	Diff line change
`@@ -146,7 +146,7 @@ public void CancelReceiveAsyncShouldStopBeforeTimeout()`
`146`	`146`	`}`
`147`	`147`
`148`	`148`	`stopwatch.Stop();`
`149`		`- stopwatch.ElapsedMilliseconds.Should().BeInRange(400, 700);`
	`149`	`+ stopwatch.ElapsedMilliseconds.Should().BeInRange(400, 711);`
`150`	`150`	`}`
`151`	`151`	`}`
`152`	`152`	`finally`
`@@ -270,6 +270,7 @@ public void SendAndReceiveLargeMessageShouldWork()`
`270`	`270`
`271`	`271`	`queue.Count().Should().Be(2);`
`272`	`272`	`queue.Transactional.Should().BeTrue();`
	`273`	`+ queue.AccessMode.Should().Be(AccessMode.SendAndReceive);`
`273`	`274`	`}`
`274`	`275`
`275`	`276`	`using (var queue = CreateLocalQueue(name, true, LocaleQueueMode.Durable))`
Original file line number	Diff line number	Diff line change
`@@ -50,6 +50,17 @@ public void SendToExistingQueueShouldNotCallCreate()`
`50`	`50`	`_messageQueueManager.Received(2).Get(".", Arg.Any<string>(), Arg.Any<bool>(), Arg.Any<QueueAccessMode>());`
`51`	`51`	`}`
`52`	`52`
	`53`	`+ [Fact]`
	`54`	`+ public void SendLargeMessageToNonTransactionalQueueShouldThrow()`
	`55`	`+ {`
	`56`	`+ SetQueue(Substitute.For<System.Messaging.MessageQueue>(), true);`
	`57`	`+`
	`58`	`+ using (var cut = CreateLocaleQueue("MyQueue", true, LocaleQueueMode.Durable, AccessMode.Send, false))`
	`59`	`+ {`
	`60`	`+ Assert.Throws<MessageSizeException>(() => cut.Send(new StringBuilder().Insert(0, "A", 5000000)));`
	`61`	`+ }`
	`62`	`+ }`
	`63`	`+`
`53`	`64`	`[Fact]`
`54`	`65`	`public void SendToNoneExistingQueueShouldCallCreate()`
`55`	`66`	`{`
`@@ -273,9 +284,9 @@ public void ReceiveOnSendQueueShouldThrowException()`
`273`	`284`	`}`
`274`	`285`	`}`
`275`	`286`
`276`		`- private IQueue CreateLocaleQueue(string queue = "MyQueue", bool privateQueue = true, LocaleQueueMode localeQueueMode = LocaleQueueMode.TemporaryMaster, AccessMode accessMode = AccessMode.SendAndReceive)`
	`287`	`+ private IQueue CreateLocaleQueue(string queue = "MyQueue", bool privateQueue = true, LocaleQueueMode localeQueueMode = LocaleQueueMode.TemporaryMaster, AccessMode accessMode = AccessMode.SendAndReceive, bool transactional = true)`
`277`	`288`	`{`
`278`		`- return new LocaleQueue(_logger, _messageQueueManager, _messageQueueTransactionFactory, queue, privateQueue, localeQueueMode, true, accessMode);`
	`289`	`+ return new LocaleQueue(_logger, _messageQueueManager, _messageQueueTransactionFactory, queue, privateQueue, localeQueueMode, transactional, accessMode);`
`279`	`290`	`}`
`280`	`291`
`281`	`292`	`private void SetQueue(System.Messaging.MessageQueue queue, bool exists)`
Original file line number	Diff line number	Diff line change
`@@ -16,7 +16,6 @@`
`16`	`16`	`using Grumpy.MessageQueue.Msmq.Interfaces;`
`17`	`17`	`using Microsoft.Extensions.Logging;`
`18`	`18`	`using Newtonsoft.Json;`
`19`		`-using ObjectExtensions = Grumpy.Common.Extensions.ObjectExtensions;`
`20`	`19`
`21`	`20`	`namespace Grumpy.MessageQueue.Msmq`
`22`	`21`	`{`
`@@ -25,7 +24,6 @@ public abstract class Queue : IQueue`
`25`	`24`	`{`
`26`	`25`	`private const int MaxMsmqMessageSize = 4096000;`
`27`	`26`	`private System.Messaging.MessageQueue _messageQueue;`
`28`		`- private readonly AccessMode _accessMode;`
`29`	`27`	`private readonly object _messageQueueLock;`
`30`	`28`	`private readonly Timer _disconnectTimer;`
`31`	`29`
`@@ -49,7 +47,7 @@ protected Queue(ILogger logger, IMessageQueueManager messageQueueManager, IMessa`
`49`	`47`	`Durable = durable;`
`50`	`48`	`_messageQueueLock = new object();`
`51`	`49`	`_disconnectTimer = new Timer(Disconnect, null, 3600000, 3600000);`
`52`		`- _accessMode = accessMode;`
	`50`	`+ AccessMode = accessMode;`
`53`	`51`	`}`
`54`	`52`
`55`	`53`	`/// <summary>`
`@@ -80,11 +78,14 @@ protected Queue(ILogger logger, IMessageQueueManager messageQueueManager, IMessa`
`80`	`78`	`/// <inheritdoc />`
`81`	`79`	`public bool Transactional { get; }`
`82`	`80`
	`81`	`+ /// <inheritdoc />`
	`82`	`+ public AccessMode AccessMode { get; }`
	`83`	`+`
`83`	`84`	`/// <inheritdoc />`
`84`	`85`	`public int Count()`
`85`	`86`	`{`
`86`		`- if (!_accessMode.In(AccessMode.Receive, AccessMode.SendAndReceive))`
`87`		`- throw new AccessModeException(nameof(Count), _accessMode);`
	`87`	`+ if (!AccessMode.In(AccessMode.Receive, AccessMode.SendAndReceive))`
	`88`	`+ throw new AccessModeException(nameof(Count), AccessMode);`
`88`	`89`
`89`	`90`	`try`
`90`	`91`	`{`
`@@ -123,16 +124,13 @@ public virtual void Connect()`
`123`	`124`	`{`
`124`	`125`	`lock (_messageQueueLock)`
`125`	`126`	`{`
`126`		`- if (_messageQueue != null)`
`127`		`- Disconnect();`
`128`		`-`
`129`	`127`	`if (_messageQueue == null)`
`130`	`128`	`{`
`131`		`- _messageQueue = GetQueue(_accessMode);`
	`129`	`+ _messageQueue = GetQueue(AccessMode);`
`132`	`130`
`133`	`131`	`if (_messageQueue != null)`
`134`	`132`	`{`
`135`		`- switch (_accessMode)`
	`133`	`+ switch (AccessMode)`
`136`	`134`	`{`
`137`	`135`	`case AccessMode.Receive:`
`138`	`136`	`_messageQueue.MessageReadPropertyFilter = new MessagePropertyFilter { AppSpecific = true, Id = true, Body = true };`
`@@ -147,11 +145,11 @@ public virtual void Connect()`
`147`	`145`	`_messageQueue.DefaultPropertiesToSend = new DefaultPropertiesToSend { Recoverable = Durable };`
`148`	`146`	`break;`
`149`	`147`	`default:`
`150`		`- throw new ArgumentOutOfRangeException(nameof(_accessMode), _accessMode, "Unknown Access Mode");`
	`148`	`+ throw new ArgumentOutOfRangeException(nameof(AccessMode), AccessMode, "Unknown Access Mode");`
`151`	`149`	`}`
`152`	`150`	`}`
`153`	`151`
`154`		`- Logger.Information("Connected to Message Queue {@Queue} {@MessageQueue}", this, _messageQueue);`
	`152`	`+ Logger.Information("Connected to Message Queue {@Queue}", this);`
`155`	`153`	`}`
`156`	`154`
`157`	`155`	`if (_messageQueue == null)`
`@@ -187,8 +185,8 @@ public void Disconnect()`
`187`	`185`	`/// <inheritdoc />`
`188`	`186`	`public void Send<T>(T message)`
`189`	`187`	`{`
`190`		`- if (!_accessMode.In(AccessMode.Send, AccessMode.SendAndReceive))`
`191`		`- throw new AccessModeException(nameof(Send), _accessMode);`
	`188`	`+ if (!AccessMode.In(AccessMode.Send, AccessMode.SendAndReceive))`
	`189`	`+ throw new AccessModeException(nameof(Send), AccessMode);`
`192`	`190`
`193`	`191`	`try`
`194`	`192`	`{`
`@@ -248,8 +246,8 @@ public async Task<ITransactionalMessage> ReceiveAsync(int millisecondsTimeout, C`
`248`	`246`
`249`	`247`	`private async Task<ITransactionalMessage> ReceiveAsyncInternal(int millisecondsTimeout, CancellationToken cancellationToken)`
`250`	`248`	`{`
`251`		`- if (!_accessMode.In(AccessMode.Receive, AccessMode.SendAndReceive))`
`252`		`- throw new AccessModeException(nameof(Send), _accessMode);`
	`249`	`+ if (!AccessMode.In(AccessMode.Receive, AccessMode.SendAndReceive))`
	`250`	`+ throw new AccessModeException(nameof(Send), AccessMode);`
`253`	`251`
`254`	`252`	`var timeout = TimeSpan.FromMilliseconds(millisecondsTimeout);`
`255`	`253`