Merge pull request #134 from Particular/release-0-2-1

Author: SeanFeldman

src/NServiceBus.Persistence.CosmosDB.PersistenceTests/ExtendedOutboxStorageTests.cs

@@ -0,0 +1,66 @@
+namespace NServiceBus.PersistenceTesting.Outbox
+{
+    using System;
+    using System.Threading.Tasks;
+    using NServiceBus.Outbox;
+    using NUnit.Framework;
+
+    [TestFixtureSource(typeof(PersistenceTestsConfiguration), "OutboxVariants")]
+    class ExtendedOutboxStorageTests
+    {
+        public ExtendedOutboxStorageTests(TestVariant param)
+        {
+            this.param = param;
+        }
+
+        [OneTimeSetUp]
+        public async Task OneTimeSetUp()
+        {
+            configuration = new PersistenceTestsConfiguration(param) { OutboxTimeToLiveInSeconds = 1 };
+            await configuration.Configure();
+        }
+
+        [OneTimeTearDown]
+        public async Task OneTimeTearDown()
+        {
+            await configuration.Cleanup();
+        }
+
+        [Test]
+        public async Task Should_expire_dispatched_messages()
+        {
+            configuration.RequiresOutboxSupport();
+
+            var storage = configuration.OutboxStorage;
+            var ctx = configuration.GetContextBagForOutbox();
+
+            var messageId = Guid.NewGuid().ToString();
+            await storage.Get(messageId, ctx);
+
+            var messageToStore = new OutboxMessage(messageId, new[] { new TransportOperation("x", null, null, null) });
+            using (var transaction = await storage.BeginTransaction(ctx))
+            {
+                await storage.Store(messageToStore, transaction, ctx);
+
+                await transaction.Commit();
+            }
+
+            await storage.SetAsDispatched(messageId, configuration.GetContextBagForOutbox());
+
+            OutboxMessage message = null;
+            for (int i = 1; i < 10; i++)
+            {
+                message = await storage.Get(messageId, configuration.GetContextBagForOutbox());
+                if (message != null)
+                {
+                    await Task.Delay(i * 550);
+                }
+            }
+
+            Assert.That(message, Is.Null, "The outbox record was not expired.");
+        }
+
+        IPersistenceTestsConfiguration configuration;
+        TestVariant param;
+    }
+}

src/NServiceBus.Persistence.CosmosDB.PersistenceTests/PersistenceTestsConfiguration.cs

@@ -32,6 +32,8 @@ public partial class PersistenceTestsConfiguration : IProvideCosmosClient
 
         public CosmosClient Client { get; } = SetupFixture.CosmosDbClient;
 
+        public int OutboxTimeToLiveInSeconds { get; set; } = 100;
+
         public Task Configure()
         {
             // with this we have a partition key per run which makes things naturally isolated
@@ -46,14 +48,17 @@ public Task Configure()
             var resolver = new ContainerHolderResolver(this, new ContainerInformation(SetupFixture.ContainerName, partitionKeyPath), SetupFixture.DatabaseName);
             SynchronizedStorage = new StorageSessionFactory(resolver, null);
             SagaStorage = new SagaPersister(serializer, false);
-            OutboxStorage = new OutboxPersister(resolver, serializer, 100);
+            OutboxStorage = new OutboxPersister(resolver, serializer, OutboxTimeToLiveInSeconds);
 
             GetContextBagForSagaStorage = () =>
             {
                 var contextBag = new ContextBag();
                 // This populates the partition key required to participate in a shared transaction
-                var setAsDispatchedHolder = new SetAsDispatchedHolder { PartitionKey = new PartitionKey(partitionKey) };
-                setAsDispatchedHolder.ContainerHolder = resolver.ResolveAndSetIfAvailable(contextBag);
+                var setAsDispatchedHolder = new SetAsDispatchedHolder
+                {
+                    PartitionKey = new PartitionKey(partitionKey),
+                    ContainerHolder = resolver.ResolveAndSetIfAvailable(contextBag)
+                };
                 contextBag.Set(setAsDispatchedHolder);
                 contextBag.Set(new PartitionKey(partitionKey));
                 return contextBag;
@@ -63,8 +68,11 @@ public Task Configure()
             {
                 var contextBag = new ContextBag();
                 // This populates the partition key required to participate in a shared transaction
-                var setAsDispatchedHolder = new SetAsDispatchedHolder { PartitionKey = new PartitionKey(partitionKey) };
-                setAsDispatchedHolder.ContainerHolder = resolver.ResolveAndSetIfAvailable(contextBag);
+                var setAsDispatchedHolder = new SetAsDispatchedHolder
+                {
+                    PartitionKey = new PartitionKey(partitionKey),
+                    ContainerHolder = resolver.ResolveAndSetIfAvailable(contextBag)
+                };
                 contextBag.Set(setAsDispatchedHolder);
                 contextBag.Set(new PartitionKey(partitionKey));
                 return contextBag;

src/NServiceBus.Persistence.CosmosDB/Installer.cs

@@ -39,7 +39,12 @@ await clientProvider.Client.CreateDatabaseIfNotExistsAsync(installerSettings.Dat
 
             var database = clientProvider.Client.GetDatabase(installerSettings.DatabaseName);
 
-            var containerProperties = new ContainerProperties(installerSettings.ContainerName, installerSettings.PartitionKeyPath);
+            var containerProperties =
+                new ContainerProperties(installerSettings.ContainerName, installerSettings.PartitionKeyPath)
+                {
+                    // in order for individual items TTL to work (example outbox records)
+                    DefaultTimeToLive = -1
+                };
 
             await database.CreateContainerIfNotExistsAsync(containerProperties)
                 .ConfigureAwait(false);