Update tests

Test/Grains/StreamConsumerGrain.cs

@@ -1,21 +1,29 @@
 using System;
 using System.Threading.Tasks;
+using Microsoft.Extensions.Logging;
 using Orleans.Providers.MongoDB.Test.GrainInterfaces;
 using Orleans.Streams;
 
 namespace Orleans.Providers.MongoDB.Test.Grains
 {
     public sealed class StreamConsumerGrain : Grain, IStreamConsumerGrain
     {
+        private readonly ILogger<StreamConsumerGrain> logger;
         private int consumedItems = 0;
 
+        public StreamConsumerGrain(ILogger<StreamConsumerGrain> logger)
+        {
+            this.logger = logger;
+        }
+
         public async Task Activate()
         {
             var streamProvider = this.GetStreamProvider("OrleansTestStream");
             var streamOfNumbers = streamProvider.GetStream<int>("MyNamespace", Guid.Empty);
 
             await streamOfNumbers.SubscribeAsync((message, token) =>
             {
+                this.logger.LogInformation("Grain {GrainId} consumed: {Message}", this.GetPrimaryKeyString(), message);
                 consumedItems++;
 
                 return Task.CompletedTask;

Test/Grains/StreamProducerGrain.cs

@@ -12,13 +12,10 @@ public async Task ProduceEvents()
             var streamProvider = this.GetStreamProvider("OrleansTestStream");
             var streamOfNumbers = streamProvider.GetStream<int>("MyNamespace", Guid.Empty);
 
-            var i = 0;
-
-            while (true)
+            for (var i = 0; i < 5; ++i)
             {
-                await streamOfNumbers.OnNextAsync(++i);
-
-                await Task.Delay(100);
+                await streamOfNumbers.OnNextAsync(i);
+                await Task.Delay(500);
             }
         }
     }

Test/Host/Program.cs

@@ -10,14 +10,12 @@
 using MongoDB.Bson.Serialization;
 using MongoDB.Bson.Serialization.Conventions;
 using MongoDB.Bson.Serialization.Serializers;
-using MongoDB.Driver;
 using Newtonsoft.Json;
 using Orleans.Configuration;
 using Orleans.Hosting;
 using Orleans.Providers.MongoDB.Configuration;
 using Orleans.Providers.MongoDB.StorageProviders.Serializers;
 using Orleans.Providers.MongoDB.Test.GrainInterfaces;
-using Orleans.Providers.MongoDB.Test.Grains;
 using Orleans.Runtime;
 
 namespace Orleans.Providers.MongoDB.Test.Host
@@ -89,8 +87,6 @@ public static async Task Main(string[] args)
 
             await host.StartAsync();
 
-            await CreateConstraints(host.Services.GetRequiredService<IMongoClient>());
-
             var clientHost = new HostBuilder()
                 .UseOrleansClient((ctx, clientBuilder) => clientBuilder
                     .UseMongoDBClient(mongoRunner.ConnectionString)
@@ -125,45 +121,26 @@ public static async Task Main(string[] args)
             await TestState(client);
             await TestStateWithCollections(client);
 
-            await TestStateConstraints(client);
-
             Console.ReadKey();
 
             await host.StopAsync();
         }
 
-        private static async Task CreateConstraints(IMongoClient mongoClient)
-        {
-            var collection = mongoClient.GetDatabase("OrleansTestApp")
-                .GetCollection<ConstrainedGrainState>("GrainsConstrainedGrain");
-
-            var indexBuilder = Builders<ConstrainedGrainState>.IndexKeys.Ascending(f => f.Name);
-
-            var options = new CreateIndexOptions
-            {
-                Unique = true
-            };
-
-            var indexModel = new CreateIndexModel<ConstrainedGrainState>(indexBuilder, options);
-
-            await collection.Indexes.CreateOneAsync(indexModel);
-        }
-
         private static async Task TestStreams(IClusterClient client)
         {
             var streamProducer = client.GetGrain<IStreamProducerGrain>(0);
-            var streamConsumer1 = client.GetGrain<IStreamConsumerGrain>(0);
-            var streamConsumer2 = client.GetGrain<IStreamConsumerGrain>(0);
+            var streamConsumer1 = client.GetGrain<IStreamConsumerGrain>(1);
+            var streamConsumer2 = client.GetGrain<IStreamConsumerGrain>(2);
 
             _ = streamProducer.ProduceEvents();
 
             await streamConsumer1.Activate();
             await streamConsumer2.Activate();
 
-            await Task.Delay(1000);
+            await Task.Delay(3000);
 
             var consumed1 = await streamConsumer1.GetConsumedItems();
-            var consumed2 = await streamConsumer1.GetConsumedItems();
+            var consumed2 = await streamConsumer2.GetConsumedItems();
 
             Console.WriteLine("Consumed Events: {0}/{1}", consumed1, consumed2);
         }
@@ -228,41 +205,6 @@ private static async Task TestState(IClusterClient client)
 
             Console.WriteLine(employeeId);
         }
-
-        private static async Task TestStateConstraints(IClusterClient client)
-        {
-            var database = client.ServiceProvider.GetRequiredService<IMongoClient>()
-                .GetDatabase("OrleansTestApp");
-
-            var guid = Guid.NewGuid().ToString();
-
-            var grain0 = client.GetGrain<IConstrainedGrain>(0);
-            await grain0.SetName(guid);
-
-            var filter = Builders<BsonDocument>.Filter.Eq("_id", "constrained/0");
-            var update = Builders<BsonDocument>.Update.Set("_etag", Guid.NewGuid().ToString());
-
-            await database.GetCollection<BsonDocument>("GrainsConstrainedGrain").UpdateOneAsync(filter, update);
-
-            try
-            {
-                await grain0.SetName(Guid.NewGuid().ToString());
-            }
-            catch (ProviderStateException ex)
-            {
-                Console.WriteLine($"Exception thrown due to invalid e-tag: {ex.Message}");
-            }
-
-            var grain1 = client.GetGrain<IConstrainedGrain>(1);
-            try
-            {
-                await grain1.SetName(guid);
-            }
-            catch (ProviderStateException ex)
-            {
-                Console.WriteLine($"Exception thrown due to unique constrain violation: {ex.Message}");
-            }
-        }
 
         private static void ApplyBsonConfiguration()
         {

UnitTest/Serializers/SerializationTests.cs

@@ -40,6 +40,35 @@ public void CorrectSerializerIsUsed()
             Assert.IsType<BinaryGrainStateSerializer>(optionsMonitor.Get("BinaryProvider").GrainStateSerializer);
         }
 
+        [Fact]
+        public void CanCustomizeJsonSerializerSettings()
+        {
+            var host = new HostBuilder()
+                .UseOrleans((ctx, siloBuilder) =>
+                {
+                    siloBuilder
+                        .UseLocalhostClustering()
+                        .AddMongoDBGrainStorage(
+                            "JsonProvider",
+                            options => options.Configure<IServiceProvider>((options, sp) =>
+                            {
+                                options.GrainStateSerializer = new JsonGrainStateSerializer(
+                                    Options.Create(new JsonGrainStateSerializerOptions
+                                    {
+                                        ConfigureJsonSerializerSettings = settings =>
+                                        {
+                                            settings.Formatting = Newtonsoft.Json.Formatting.Indented;
+                                        }
+                                    }),
+                                    sp);
+                            }));
+                })
+                .Build();
+
+            var optionsMonitor = host.Services.GetRequiredService<IOptionsMonitor<MongoDBGrainStorageOptions>>();
+            Assert.IsType<JsonGrainStateSerializer>(optionsMonitor.Get("JsonProvider").GrainStateSerializer);
+        }
+
         [Fact]
         public async void BsonSerializerForPubSubStore_Throws()
         {

UnitTest/Storage/StorageTests.cs

@@ -0,0 +1,67 @@
+using System;
+using System.Threading.Tasks;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Hosting;
+using MongoDB.Bson;
+using MongoDB.Driver;
+using Orleans.Hosting;
+using Orleans.Providers.MongoDB.StorageProviders.Serializers;
+using Orleans.Providers.MongoDB.UnitTest.Fixtures;
+using Orleans.Runtime;
+using TestExtensions;
+using Xunit;
+using static Orleans.Providers.MongoDB.UnitTest.Storage.TestGrains.StorageTests;
+
+namespace Orleans.Providers.MongoDB.UnitTest.Storage
+{
+    [Collection(TestEnvironmentFixture.DefaultCollection)]
+    public partial class StorageTests
+    {
+        [Fact]
+        public async Task ThrowErrorWhenUpdatingCollectionWithAUniqueIndex()
+        {
+            var host = new HostBuilder()
+                .UseOrleans((ctx, siloBuilder) =>
+                {
+                    siloBuilder.Services
+                        .AddSingletonNamedService<IGrainStateSerializer, BsonGrainStateSerializer>(ProviderConstants.DEFAULT_PUBSUB_PROVIDER_NAME);
+
+                    siloBuilder
+                        .AddMemoryStreams<DefaultMemoryMessageBodySerializer>("OrleansTestStream")
+                        .UseLocalhostClustering()
+                        .UseMongoDBClient(MongoDatabaseFixture.DatabaseConnectionString)
+                        .AddMongoDBGrainStorageAsDefault(options => options.DatabaseName = "OrleansTestApp")
+                        .AddMongoDBGrainStorage(ProviderConstants.DEFAULT_PUBSUB_PROVIDER_NAME, options => options.DatabaseName = "OrleansTestApp");
+                })
+                .Build();
+
+            await host.StartAsync();
+
+            var client = host.Services.GetRequiredService<IClusterClient>();
+            var mongoClient = client.ServiceProvider.GetRequiredService<IMongoClient>();
+            var database = mongoClient.GetDatabase("OrleansTestApp");
+            var collection = database.GetCollection<ConstrainedGrainState>("GrainsConstrainedGrain");
+
+            await collection.Indexes.CreateOneAsync(
+                new CreateIndexModel<ConstrainedGrainState>(
+                    Builders<ConstrainedGrainState>.IndexKeys.Ascending(f => f.Name),
+                    new CreateIndexOptions { Unique = true }));
+
+            var guid = Guid.NewGuid().ToString();
+
+            var grain0 = client.GetGrain<IConstrainedGrain>(0);
+            await grain0.SetName(guid);
+
+            await database.GetCollection<BsonDocument>("GrainsConstrainedGrain").UpdateOneAsync(
+                    Builders<BsonDocument>.Filter.Eq("_id", "constrained/0"),
+                    Builders<BsonDocument>.Update.Set("_etag", Guid.NewGuid().ToString()));
+
+            var exception0 = await Assert.ThrowsAsync<ProviderStateException>(async () => await grain0.SetName(Guid.NewGuid().ToString()));
+            Assert.Equal("A write operation resulted in an error. WriteError: { Category : \"DuplicateKey\", Code : 11000, Message : \"E11000 duplicate key error collection: OrleansTestApp.GrainsConstrainedGrain index: _id_ dup key: { _id: \"constrained/0\" }\" }.", exception0.Message);
+
+            var grain1 = client.GetGrain<IConstrainedGrain>(1);
+            var exception1 = await Assert.ThrowsAsync<ProviderStateException>(async () => await grain1.SetName(guid));
+            Assert.Contains("A write operation resulted in an error. WriteError: { Category : \"DuplicateKey\", Code : 11000, Message : \"E11000 duplicate key error collection: OrleansTestApp.GrainsConstrainedGrain index: Name_1 dup key: { Name: null }\" }.", exception1.Message);
+        }
+    }
+}

UnitTest/Storage/TestGrains/ConstrainedGrain.cs

@@ -0,0 +1,33 @@
+using System;
+using System.Threading.Tasks;
+using Orleans.Runtime;
+
+namespace Orleans.Providers.MongoDB.UnitTest.Storage.TestGrains
+{
+    public partial class StorageTests
+    {
+        public class ConstrainedGrain : Grain, IConstrainedGrain
+        {
+            private readonly IPersistentState<ConstrainedGrainState> state;
+
+            public ConstrainedGrain([PersistentState(nameof(ConstrainedGrain))] IPersistentState<ConstrainedGrainState> state)
+            {
+                this.state = state;
+            }
+
+            public async Task SetName(string name)
+            {
+                state.State.Name = name;
+
+                try
+                {
+                    await state.WriteStateAsync();
+                }
+                catch (Exception ex)
+                {
+                    throw new ProviderStateException(ex.Message);
+                }
+            }
+        }
+    }
+}

UnitTest/Storage/TestGrains/ConstrainedGrainState.cs

@@ -0,0 +1,10 @@
+namespace Orleans.Providers.MongoDB.UnitTest.Storage.TestGrains
+{
+    public partial class StorageTests
+    {
+        public class ConstrainedGrainState
+        {
+            public string Name { get; set; }
+        }
+    }
+}

UnitTest/Storage/TestGrains/IConstrainedGrain.cs

@@ -0,0 +1,12 @@
+using System.Threading.Tasks;
+
+namespace Orleans.Providers.MongoDB.UnitTest.Storage.TestGrains
+{
+    public partial class StorageTests
+    {
+        public interface IConstrainedGrain : IGrainWithIntegerKey
+        {
+            Task SetName(string name);
+        }
+    }
+}