Merge pull request #63 from Arkatufus/master

Version 0.3.2 Release

RELEASE_NOTES.md

@@ -1,5 +1,7 @@
-## [0.3.0] / 24 May 2022
-- [Add interfaces for the `ActorRegistry` to allow mocking in tests](https://github.com/akkadotnet/Akka.Hosting/pull/42)
-- [Fixed: Akka.Hosting NRE upon shutdown](https://github.com/akkadotnet/Akka.Hosting/pull/49)
-- [added throw-able `ActorRegistry.Register` method](https://github.com/akkadotnet/Akka.Hosting/pull/50)
-- [Akka.Cluster.Hosting: adding `ClusterSingleton` hosting methods](https://github.com/akkadotnet/Akka.Hosting/pull/51)
+## [0.3.2] / 13 June 2022
+- [Fixed: WithDistributedPubSub role HOCON settings not inserted in proper order](https://github.com/akkadotnet/Akka.Hosting/issues/60)
+
+## [0.3.1] / 09 June 2022
+- [Fixed: WithDistributedPubSub throws NullReferenceException](https://github.com/akkadotnet/Akka.Hosting/issues/55)
+- [Introduced `AddHoconFile` method](https://github.com/akkadotnet/Akka.Hosting/pull/58)
+- [Upgraded to Akka.NET 1.4.39](https://github.com/akkadotnet/akka.net/releases/tag/1.4.39)

src/Akka.Cluster.Hosting.Tests/DistributedPubSubSpecs.cs

@@ -0,0 +1,126 @@
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+using Akka.Actor;
+using Akka.Cluster.Tools.PublishSubscribe;
+using Akka.Event;
+using Akka.Hosting;
+using Akka.Remote.Hosting;
+using FluentAssertions;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Hosting;
+using Microsoft.Extensions.Logging;
+using Xunit;
+using Xunit.Abstractions;
+using LogLevel = Microsoft.Extensions.Logging.LogLevel;
+
+namespace Akka.Cluster.Hosting.Tests;
+
+public class DistributedPubSubSpecs : IAsyncLifetime
+{
+    private readonly ITestOutputHelper _helper;
+    private readonly Action<AkkaConfigurationBuilder> _specBuilder;
+    private readonly ClusterOptions _clusterOptions;    
+    private IHost _host;
+    private ActorSystem _system;
+    private ILoggingAdapter _log;
+    private Cluster _cluster;
+    private TestKit.Xunit2.TestKit _testKit;
+
+    private IActorRef _mediator;
+
+    public DistributedPubSubSpecs(ITestOutputHelper helper)
+    {
+        _helper = helper;
+        _specBuilder = _ => { };
+        _clusterOptions = new ClusterOptions { Roles = new[] { "my-host" } };
+    }
+
+    // Issue #55 https://github.com/akkadotnet/Akka.Hosting/issues/55
+    [Fact]
+    public Task Should_launch_distributed_pub_sub_with_roles()
+    {
+        var testProbe = _testKit.CreateTestProbe(_system);
+
+        // act
+        testProbe.Send(_mediator, new Subscribe("testSub", testProbe));
+        var response = testProbe.ExpectMsg<SubscribeAck>();
+
+        // assert
+        _system.Settings.Config.GetString("akka.cluster.pub-sub.role").Should().Be("my-host");
+        response.Subscribe.Topic.Should().Be("testSub");
+        response.Subscribe.Ref.Should().Be(testProbe);
+
+        return Task.CompletedTask;
+    }
+
+    [Fact]
+    public Task Distributed_pub_sub_should_work()
+    {
+        const string topic = "testSub";
+
+        var subscriber = _testKit.CreateTestProbe(_system);
+        var publisher = _testKit.CreateTestProbe(_system);
+
+        subscriber.Send(_mediator, new Subscribe(topic, subscriber));
+        subscriber.ExpectMsg<SubscribeAck>();
+
+        publisher.Send(_mediator, new Publish(topic, "test message"));
+        subscriber.ExpectMsg("test message");
+
+        return Task.CompletedTask;
+    }
+
+    public async Task InitializeAsync()
+    {
+        using var cancellationTokenSource = new CancellationTokenSource(TimeSpan.FromSeconds(10));
+
+        _host = new HostBuilder()
+            .ConfigureLogging(builder =>
+            {
+                builder.AddProvider(new XUnitLoggerProvider(_helper, LogLevel.Information));
+            })
+            .ConfigureServices(collection =>
+            {
+                collection
+                    .AddAkka("TestSys", (configurationBuilder, _) =>
+                    {
+                        configurationBuilder
+                            .AddHocon(TestKit.Xunit2.TestKit.DefaultConfig)
+                            .WithRemoting("localhost", 0)
+                            .WithClustering(_clusterOptions)
+                            .WithActors((system, _) =>
+                            {
+                                _testKit = new TestKit.Xunit2.TestKit(system, _helper);
+                                _system = system;
+                                _log = Logging.GetLogger(system, this);
+                                _cluster = Cluster.Get(system);
+
+                                _log.Info("Distributed pub-sub test system initialized.");
+                            })
+                            .WithDistributedPubSub("my-host");
+                        _specBuilder(configurationBuilder);
+                    });
+            }).Build();
+
+        await _host.StartAsync(cancellationTokenSource.Token);
+
+        // Lifetime should be healthy
+        var lifetime = _host.Services.GetRequiredService<IHostApplicationLifetime>();
+        lifetime.ApplicationStopped.IsCancellationRequested.Should().BeFalse();
+        lifetime.ApplicationStopping.IsCancellationRequested.Should().BeFalse();
+
+        // Join cluster
+        var myAddress = _cluster.SelfAddress;
+        await _cluster.JoinAsync(myAddress, cancellationTokenSource.Token); // force system to wait until we're up
+
+        // Prepare test
+        var registry = _host.Services.GetRequiredService<ActorRegistry>();
+        _mediator = registry.Get<DistributedPubSub>();
+    }
+
+    public async Task DisposeAsync()
+    {
+        await _host.StopAsync();
+    }
+}

src/Akka.Cluster.Hosting.Tests/XUnitLogger.cs

@@ -0,0 +1,84 @@
+using System;
+using Microsoft.Extensions.Logging;
+using Xunit.Abstractions;
+
+namespace Akka.Cluster.Hosting.Tests;
+
+public class XUnitLogger: ILogger
+{
+    private const string NullFormatted = "[null]";
+
+    private readonly string _category;
+    private readonly ITestOutputHelper _helper;
+    private readonly LogLevel _logLevel;
+
+    public XUnitLogger(string category, ITestOutputHelper helper, LogLevel logLevel)
+    {
+        _category = category;
+        _helper = helper;
+        _logLevel = logLevel;
+    }
+
+    public void Log<TState>(LogLevel logLevel, EventId eventId, TState state, Exception exception, Func<TState, Exception, string> formatter)
+    {
+        if (!IsEnabled(logLevel))
+            return;
+
+        if (!TryFormatMessage(state, exception, formatter, out var formattedMessage))
+            return;
+
+        WriteLogEntry(logLevel, eventId, formattedMessage, exception);
+    }
+
+    private void WriteLogEntry(LogLevel logLevel, EventId eventId, string message, Exception exception)
+    {
+        var level = logLevel switch
+        {
+            LogLevel.Critical => "CRT",
+            LogLevel.Debug => "DBG",
+            LogLevel.Error => "ERR",
+            LogLevel.Information => "INF",
+            LogLevel.Warning => "WRN",
+            LogLevel.Trace => "DBG",
+            _ => "???"
+        };
+
+        var msg = $"{DateTime.Now}:{level}:{_category}:{eventId} {message}";
+        if (exception != null)
+            msg += $"\n{exception.GetType()} {exception.Message}\n{exception.StackTrace}";
+        _helper.WriteLine(msg);
+    }
+
+    public bool IsEnabled(LogLevel logLevel)
+    {
+        return logLevel switch
+        {
+            LogLevel.None => false,
+            _ => logLevel >= _logLevel
+        };
+    }
+
+    public IDisposable BeginScope<TState>(TState state)
+    {
+        throw new NotImplementedException();
+    }
+
+    private static bool TryFormatMessage<TState>(
+        TState state,
+        Exception exception,
+        Func<TState, Exception, string> formatter,
+        out string result)
+    {
+        formatter = formatter ?? throw new ArgumentNullException(nameof(formatter));
+
+        var formattedMessage = formatter(state, exception);
+        if (formattedMessage == NullFormatted)
+        {
+            result = null;
+            return false;
+        }
+
+        result = formattedMessage;
+        return true;
+    }
+}

src/Akka.Cluster.Hosting.Tests/XUnitLoggerProvider.cs

@@ -0,0 +1,26 @@
+using Microsoft.Extensions.Logging;
+using Xunit.Abstractions;
+
+namespace Akka.Cluster.Hosting.Tests;
+
+public class XUnitLoggerProvider : ILoggerProvider
+{
+    private readonly ITestOutputHelper _helper;
+    private readonly LogLevel _logLevel;
+
+    public XUnitLoggerProvider(ITestOutputHelper helper, LogLevel logLevel)
+    {
+        _helper = helper;
+        _logLevel = logLevel;
+    }
+
+    public void Dispose()
+    {
+        // no-op
+    }
+
+    public ILogger CreateLogger(string categoryName)
+    {
+        return new XUnitLogger(categoryName, _helper, _logLevel);
+    }
+}

src/Akka.Cluster.Hosting/AkkaClusterHostingExtensions.cs

@@ -254,7 +254,7 @@ public static AkkaConfigurationBuilder WithDistributedPubSub(this AkkaConfigurat
             var middle = builder.AddHocon(DistributedPubSub.DefaultConfig());
             if (!string.IsNullOrEmpty(role)) // add role config
             {
-                middle = middle.AddHocon($"akka.cluster.pub-sub = \"{role}\"");
+                middle = middle.AddHocon($"akka.cluster.pub-sub.role = \"{role}\"", HoconAddMode.Prepend);
             }
 
             return middle.WithActors((system, registry) =>

src/Akka.Hosting.Tests/Akka.Hosting.Tests.csproj

@@ -13,4 +13,11 @@
   <ItemGroup>
     <ProjectReference Include="..\Akka.Hosting\Akka.Hosting.csproj" />
   </ItemGroup>
+
+  <ItemGroup>
+    <None Remove="test.hocon" />
+    <Content Include="test.hocon">
+      <CopyToOutputDirectory>Always</CopyToOutputDirectory>
+    </Content>
+  </ItemGroup>
 </Project>

src/Akka.Hosting.Tests/DISanityCheckSpecs.cs

@@ -5,12 +5,12 @@
 using Akka.DependencyInjection;
 using FluentAssertions;
 using Microsoft.Extensions.DependencyInjection;
-using Microsoft.Extensions.Hosting;
 using Xunit;
+using static Akka.Hosting.Tests.TestHelpers;
 
 namespace Akka.Hosting.Tests;
 
-public class DiSanityCheckSpecs
+public class DiSanityCheckSpecs 
 {
     public interface IMySingletonInterface{}
 
@@ -36,19 +36,6 @@ public SingletonActor(IMySingletonInterface singleton)
             });
         }
     }
-
-    private static async Task<IHost> StartHost(Action<IServiceCollection> testSetup)
-    {
-        var host = new HostBuilder()
-        .ConfigureServices(services =>
-        {
-            services.AddSingleton<IMySingletonInterface, MySingletonImpl>();
-            testSetup(services);
-        }).Build();
-
-        await host.StartAsync();
-        return host;
-    }
 
     /// <summary>
     /// Sanity check: things registered as singletons prior to the creation of the <see cref="ActorSystem"/> should

src/Akka.Hosting.Tests/HoconSpecs.cs

@@ -0,0 +1,28 @@
+using System.Threading.Tasks;
+using Akka.Actor;
+using FluentAssertions;
+using Microsoft.Extensions.DependencyInjection;
+using Xunit;
+using static Akka.Hosting.Tests.TestHelpers;
+
+namespace Akka.Hosting.Tests;
+
+public class HoconSpecs
+{
+    [Fact]
+    public async Task Should_load_HOCON_from_file()
+    {
+        // arrange
+        using var host = await StartHost(collection => collection.AddAkka("Test", builder =>
+        {
+            builder.AddHoconFile("test.hocon");
+        }));
+
+        // act
+        var sys = host.Services.GetRequiredService<ActorSystem>();
+        var hocon = sys.Settings.Config;
+
+        // assert
+        hocon.HasPath("petabridge.cmd").Should().BeTrue();
+    }
+}

src/Akka.Hosting.Tests/TestHelpers.cs

@@ -0,0 +1,22 @@
+using System;
+using System.Threading.Tasks;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Hosting;
+
+namespace Akka.Hosting.Tests;
+
+public static class TestHelpers
+{
+    public static async Task<IHost> StartHost(Action<IServiceCollection> testSetup)
+    {
+        var host = new HostBuilder()
+            .ConfigureServices(services =>
+            {
+                services.AddSingleton<DiSanityCheckSpecs.IMySingletonInterface, DiSanityCheckSpecs.MySingletonImpl>();
+                testSetup(services);
+            }).Build();
+
+        await host.StartAsync();
+        return host;
+    }
+}

src/Akka.Hosting.Tests/test.hocon

@@ -0,0 +1,9 @@
+# See petabridge.cmd configuration options here: https://cmd.petabridge.com/articles/install/host-configuration.html
+petabridge.cmd{
+    # default IP address used to listen for incoming petabridge.cmd client connections
+    # should be a safe default as it listens on "all network interfaces".
+    host = "0.0.0.0"
+
+    # default port number used to listen for incoming petabridge.cmd client connections
+    port = 9110
+}