feat: add local redis test

.dockerignore

@@ -0,0 +1,25 @@
+**/.classpath
+**/.dockerignore
+**/.env
+**/.git
+**/.gitignore
+**/.project
+**/.settings
+**/.toolstarget
+**/.vs
+**/.vscode
+**/*.*proj.user
+**/*.dbmdl
+**/*.jfm
+**/azds.yaml
+**/bin
+**/charts
+**/docker-compose*
+**/Dockerfile*
+**/node_modules
+**/npm-debug.log
+**/obj
+**/secrets.dev.yaml
+**/values.dev.yaml
+LICENSE
+README.md

docker-compose.yml

@@ -0,0 +1,7 @@
+version: "3.4"
+
+services:
+  redis:
+    image: "redis:6.2.1"
+    ports:
+      - "6379:6379"

tests/MessagePipe.Tests/MessagePipe.Tests.csproj

@@ -1,4 +1,4 @@
-<Project Sdk="Microsoft.NET.Sdk">
+<Project Sdk="Microsoft.NET.Sdk">
 
     <PropertyGroup>
         <TargetFramework>net5.0</TargetFramework>
@@ -9,7 +9,8 @@
         <PackageReference Include="FluentAssertions" Version="5.10.3" />
         <PackageReference Include="Microsoft.Extensions.DependencyInjection" Version="5.0.1" />
         <PackageReference Include="Microsoft.NET.Test.Sdk" Version="16.9.4" />
-        <PackageReference Include="xunit" Version="2.4.1" />
+		<PackageReference Include="StackExchange.Redis" Version="2.2.4" />
+		<PackageReference Include="xunit" Version="2.4.1" />
         <PackageReference Include="xunit.runner.visualstudio" Version="2.4.3">
             <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
             <PrivateAssets>all</PrivateAssets>

tests/MessagePipe.Tests/RedisPubSubKeyedASync.cs

@@ -0,0 +1,67 @@
+using FluentAssertions;
+using MessagePipe;
+using MessagePipe.Tests;
+using Microsoft.Extensions.DependencyInjection;
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading;
+using System.Threading.Tasks;
+using Xunit;
+
+// for check diagnostics, modify namespace.
+namespace __MessagePipe.Tests
+{
+    public class RedisPubSubKeyedASync
+    {
+        [Fact]
+        public async Task SimplePush()
+        {
+            const string Key1 = "foo";
+            const string Key2 = "bar";
+
+            var conection = await StackExchange.Redis.ConnectionMultiplexer.ConnectAsync("localhost");
+            var provider = TestHelper.BuildRedisServiceProvider(conection);
+
+            var info = provider.GetRequiredService<MessagePipeDiagnosticsInfo>();
+            var p = provider.GetRequiredService<IAsyncPublisher<string, string>>();
+            var s = provider.GetRequiredService<IAsyncSubscriber<string, string>>();
+
+            var result = new List<string>();
+            var d1 = s.Subscribe(Key1, async (x, ct) => result.Add("1:" + x));
+            var d2 = s.Subscribe(Key2, async (x, ct) => result.Add("2:" + x));
+            var d3 = s.Subscribe(Key1, async (x, ct) => result.Add("3:" + x));
+
+            info.SubscribeCount.Should().Be(3);
+
+            // use BeEquivalentTo, allow different order
+
+            p.Publish(Key1, "one");
+            result.Should().BeEquivalentTo("1:one", "3:one");
+            result.Clear();
+
+            p.Publish(Key2, "one");
+            result.Should().BeEquivalentTo("2:one");
+            result.Clear();
+
+            d3.Dispose();
+
+            p.Publish(Key1, "two");
+            result.Should().BeEquivalentTo("1:two");
+            result.Clear();
+
+            d1.Dispose();
+            d2.Dispose();
+
+            p.Publish(Key1, "zero");
+            p.Publish(Key2, "zero");
+
+            result.Should().Equal();
+            result.Clear();
+
+            info.SubscribeCount.Should().Be(0);
+        }
+
+    }
+}

tests/MessagePipe.Tests/RedisPubSubKeyedSync.cs

@@ -0,0 +1,67 @@
+using FluentAssertions;
+using MessagePipe;
+using MessagePipe.Tests;
+using Microsoft.Extensions.DependencyInjection;
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading;
+using System.Threading.Tasks;
+using Xunit;
+
+// for check diagnostics, modify namespace.
+namespace __MessagePipe.Tests
+{
+    public class RedisPubSubKeyedSync
+    {
+        [Fact]
+        public async Task SimplePush()
+        {
+            const string Key1 = "foo";
+            const string Key2 = "bar";
+
+            var conection = await StackExchange.Redis.ConnectionMultiplexer.ConnectAsync("localhost");
+            var provider = TestHelper.BuildRedisServiceProvider(conection);
+
+            var info = provider.GetRequiredService<MessagePipeDiagnosticsInfo>();
+            var p = provider.GetRequiredService<IPublisher<string, string>>();
+            var s = provider.GetRequiredService<ISubscriber<string, string>>();
+
+            var result = new List<string>();
+            var d1 = s.Subscribe(Key1, x => result.Add("1:" + x));
+            var d2 = s.Subscribe(Key2, x => result.Add("2:" + x));
+            var d3 = s.Subscribe(Key1, x => result.Add("3:" + x));
+
+            info.SubscribeCount.Should().Be(3);
+
+            // use BeEquivalentTo, allow different order
+
+            p.Publish(Key1, "one");
+            result.Should().BeEquivalentTo("1:one", "3:one");
+            result.Clear();
+
+            p.Publish(Key2, "one");
+            result.Should().BeEquivalentTo("2:one");
+            result.Clear();
+
+            d3.Dispose();
+
+            p.Publish(Key1, "two");
+            result.Should().BeEquivalentTo("1:two");
+            result.Clear();
+
+            d1.Dispose();
+            d2.Dispose();
+
+            p.Publish(Key1, "zero");
+            p.Publish(Key2, "zero");
+
+            result.Should().Equal();
+            result.Clear();
+
+            info.SubscribeCount.Should().Be(0);
+        }
+
+    }
+}

tests/MessagePipe.Tests/RedisPubsubKeylessAsync.cs

@@ -0,0 +1,91 @@
+using FluentAssertions;
+using MessagePipe;
+using MessagePipe.Tests;
+using Microsoft.Extensions.DependencyInjection;
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading;
+using System.Threading.Tasks;
+using Xunit;
+
+// for check diagnostics, modify namespace.
+namespace __MessagePipe.Tests
+{
+    public class RedisPubsubKeylessAsync
+    {
+        [Fact]
+        public async Task SameAsSync()
+        {
+            var conection = await StackExchange.Redis.ConnectionMultiplexer.ConnectAsync("localhost");
+            var provider = TestHelper.BuildRedisServiceProvider(conection);
+
+            var info = provider.GetRequiredService<MessagePipeDiagnosticsInfo>();
+            var p = provider.GetRequiredService<IAsyncPublisher<string>>();
+            var s = provider.GetRequiredService<IAsyncSubscriber<string>>();
+
+            var result = new List<string>();
+            var d1 = s.Subscribe(async (x, c) => result.Add("1:" + x));
+            var d2 = s.Subscribe(async (x, c) => result.Add("2:" + x));
+            var d3 = s.Subscribe(async (x, c) => result.Add("3:" + x));
+
+            info.SubscribeCount.Should().Be(3);
+
+            // use BeEquivalentTo, allow different order
+
+            p.Publish("one");
+            result.Should().BeEquivalentTo("1:one", "2:one", "3:one");
+            result.Clear();
+
+            p.Publish("one");
+            result.Should().BeEquivalentTo("1:one", "2:one", "3:one");
+            result.Clear();
+
+            d2.Dispose();
+
+            p.Publish("two");
+            result.Should().BeEquivalentTo("1:two", "3:two");
+            result.Clear();
+
+            d3.Dispose();
+            p.Publish("three");
+            result.Should().BeEquivalentTo("1:three");
+            result.Clear();
+
+            d1.Dispose();
+            result.Should().Equal();
+            result.Clear();
+
+            info.SubscribeCount.Should().Be(0);
+        }
+
+        [Fact]
+        public async Task Async()
+        {
+            var conection = await StackExchange.Redis.ConnectionMultiplexer.ConnectAsync("localhost");
+            var provider = TestHelper.BuildRedisServiceProvider(conection);
+
+            var p = provider.GetRequiredService<IAsyncPublisher<string>>();
+            var s = provider.GetRequiredService<IAsyncSubscriber<string>>();
+
+            var result = new List<string>();
+
+            s.Subscribe(async (x, ct) =>
+            {
+                await Task.Delay(TimeSpan.FromSeconds(3), ct);
+                Console.WriteLine("OK?");
+            });
+
+            s.Subscribe(async (x, ct) =>
+            {
+                await Task.Delay(TimeSpan.FromSeconds(5), ct);
+                Console.WriteLine("OK2?");
+            });
+
+            // TODO:calclate time
+            var cts = new CancellationTokenSource();
+            await p.PublishAsync("takoyaki", AsyncPublishStrategy.Sequential, cts.Token);
+        }
+    }
+}

tests/MessagePipe.Tests/RedisPubsubKeylessSync.cs

@@ -0,0 +1,63 @@
+using FluentAssertions;
+using MessagePipe;
+using MessagePipe.Tests;
+using Microsoft.Extensions.DependencyInjection;
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading;
+using System.Threading.Tasks;
+using Xunit;
+
+// for check diagnostics, modify namespace.
+namespace __MessagePipe.Tests
+{
+    public class RedisPubsubKeylessSync
+    {
+        [Fact]
+        public async Task SimplePush()
+        {
+            var conection = await StackExchange.Redis.ConnectionMultiplexer.ConnectAsync("localhost");
+            var provider = TestHelper.BuildRedisServiceProvider(conection);
+
+            var info = provider.GetRequiredService<MessagePipeDiagnosticsInfo>();
+            var p = provider.GetRequiredService<IPublisher<string>>();
+            var s = provider.GetRequiredService<ISubscriber<string>>();
+
+            var result = new List<string>();
+            var d1 = s.Subscribe(x => result.Add("1:" + x));
+            var d2 = s.Subscribe(x => result.Add("2:" + x));
+            var d3 = s.Subscribe(x => result.Add("3:" + x));
+
+            info.SubscribeCount.Should().Be(3);
+
+            // use BeEquivalentTo, allow different order
+
+            p.Publish("one");
+            result.Should().BeEquivalentTo("1:one", "2:one", "3:one");
+            result.Clear();
+
+            p.Publish("one");
+            result.Should().BeEquivalentTo("1:one", "2:one", "3:one");
+            result.Clear();
+
+            d2.Dispose();
+
+            p.Publish("two");
+            result.Should().BeEquivalentTo("1:two", "3:two");
+            result.Clear();
+
+            d3.Dispose();
+            p.Publish("three");
+            result.Should().BeEquivalentTo("1:three");
+            result.Clear();
+
+            d1.Dispose();
+            result.Should().Equal();
+            result.Clear();
+
+            info.SubscribeCount.Should().Be(0);
+        }
+    }
+}