Skip to content
Original file line number Diff line number Diff line change
Expand Up @@ -7,8 +7,8 @@

#nullable enable
using System;
using System.Linq;
using System.Threading.Tasks;
using Akka.Actor;
using Akka.Cluster.Metrics.Serialization;
using Akka.Configuration;
using Akka.Serialization;
Expand Down Expand Up @@ -50,9 +50,13 @@ public MetricsSerializersSpec(ITestOutputHelper output) : base(MetricsRows, outp
public void Should_have_a_reference_conf_row_When_the_table_lists_a_type()
{
var table = new MetricsSerializers();
ModuleSerializerSpecs.AssertTableMatchesConfig(MetricsRows, table.Serializers.Select(s => s.Type), table.BoundTypes);
ModuleSerializerSpecs.AssertTableMatchesConfig(MetricsRows, table.Create((ExtendedActorSystem)Sys));
}

[Fact(DisplayName = "MetricsSerializers should build without throwing on a system that never loaded its reference.conf")]
public async Task Should_build_without_throwing_When_its_config_is_absent()
=> await ModuleSerializerSpecs.AssertBuildsWithoutModuleConfig("metrics-no-config", s => new MetricsSerializers().Create(s));

[Fact(DisplayName = "Serialization should resolve reference.conf rows spelled as Akka.Hosting writes them when dynamic type loading is off")]
public async Task Should_resolve_assembly_qualified_names_When_dynamic_type_loading_is_disabled()
=> await ModuleSerializerSpecs.AssertHostingSpellingResolves("metrics-aqn", MetricsRows, Sys);
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -6,8 +6,8 @@
//-----------------------------------------------------------------------

#nullable enable
using System;
using System.Collections.Generic;
using System.Collections.Immutable;
using Akka.Actor;
using Akka.Serialization;

namespace Akka.Cluster.Metrics.Serialization
Expand All @@ -17,19 +17,14 @@ namespace Akka.Cluster.Metrics.Serialization
/// </summary>
internal sealed class MetricsSerializers : ModuleSerializers
{
// the constructor reflection picks for reference.conf; ClusterMetricsMessageSerializer has one, so it always gets that one
public override IReadOnlyList<ModuleSerializer> Serializers { get; } = new[]
{
new ModuleSerializer(typeof(ClusterMetricsMessageSerializer), (system, _) => new ClusterMetricsMessageSerializer(system)),
};

public override IReadOnlyList<Type> BoundTypes { get; } = new[]
{
typeof(MetricsGossipEnvelope),
typeof(AdaptiveLoadBalancingPool),
typeof(MixMetricsSelector),
typeof(CpuMetricsSelector),
typeof(MemoryMetricsSelector),
};
public override ImmutableHashSet<SerializerDetails> Create(ExtendedActorSystem system) => ImmutableHashSet.Create(
SerializerDetails.Create("akka-cluster-metrics", new ClusterMetricsMessageSerializer(system),
ImmutableHashSet.Create(
typeof(MetricsGossipEnvelope),
typeof(AdaptiveLoadBalancingPool),
typeof(MixMetricsSelector),
typeof(CpuMetricsSelector),
typeof(MemoryMetricsSelector)))
);
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -6,8 +6,8 @@
//-----------------------------------------------------------------------

#nullable enable
using System.Linq;
using System.Threading.Tasks;
using Akka.Actor;
using Akka.Cluster.Sharding.Serialization;
using Akka.Configuration;
using Akka.Serialization;
Expand Down Expand Up @@ -45,9 +45,13 @@ public ShardingSerializersSpec(ITestOutputHelper output) : base(ClusterSharding.
public void Should_have_a_reference_conf_row_When_the_table_lists_a_type()
{
var table = new ShardingSerializers();
ModuleSerializerSpecs.AssertTableMatchesConfig(ShardingRows, table.Serializers.Select(s => s.Type), table.BoundTypes);
ModuleSerializerSpecs.AssertTableMatchesConfig(ShardingRows, table.Create((ExtendedActorSystem)Sys));
}

[Fact(DisplayName = "ShardingSerializers should build without throwing on a system that never loaded its reference.conf")]
public async Task Should_build_without_throwing_When_its_config_is_absent()
=> await ModuleSerializerSpecs.AssertBuildsWithoutModuleConfig("sharding-no-config", s => new ShardingSerializers().Create(s));

[Fact(DisplayName = "Serialization should resolve reference.conf rows spelled as Akka.Hosting writes them when dynamic type loading is off")]
public async Task Should_resolve_assembly_qualified_names_When_dynamic_type_loading_is_disabled()
=> await ModuleSerializerSpecs.AssertHostingSpellingResolves("sharding-aqn", ShardingRows, Sys);
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -6,8 +6,8 @@
//-----------------------------------------------------------------------

#nullable enable
using System;
using System.Collections.Generic;
using System.Collections.Immutable;
using Akka.Actor;
using Akka.Serialization;

namespace Akka.Cluster.Sharding.Serialization
Expand All @@ -17,15 +17,9 @@ namespace Akka.Cluster.Sharding.Serialization
/// </summary>
internal sealed class ShardingSerializers : ModuleSerializers
{
// the constructor reflection picks for reference.conf; ClusterShardingMessageSerializer has one, so it always gets that one
public override IReadOnlyList<ModuleSerializer> Serializers { get; } = new[]
{
new ModuleSerializer(typeof(ClusterShardingMessageSerializer), (system, _) => new ClusterShardingMessageSerializer(system)),
};

public override IReadOnlyList<Type> BoundTypes { get; } = new[]
{
typeof(IClusterShardingSerializable),
};
public override ImmutableHashSet<SerializerDetails> Create(ExtendedActorSystem system) => ImmutableHashSet.Create(
SerializerDetails.Create("akka-sharding", new ClusterShardingMessageSerializer(system),
ImmutableHashSet.Create(typeof(IClusterShardingSerializable)))
);
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -6,8 +6,8 @@
//-----------------------------------------------------------------------

#nullable enable
using System.Linq;
using System.Threading.Tasks;
using Akka.Actor;
using Akka.Cluster.Tools.Client;
using Akka.Cluster.Tools.Client.Serialization;
using Akka.Cluster.Tools.PublishSubscribe;
Expand Down Expand Up @@ -51,9 +51,13 @@ public ToolsSerializersSpec(ITestOutputHelper output) : base(ToolsRows, output)
public void Should_have_a_reference_conf_row_When_the_table_lists_a_type()
{
var table = new ToolsSerializers();
ModuleSerializerSpecs.AssertTableMatchesConfig(ToolsRows, table.Serializers.Select(s => s.Type), table.BoundTypes);
ModuleSerializerSpecs.AssertTableMatchesConfig(ToolsRows, table.Create((ExtendedActorSystem)Sys));
}

[Fact(DisplayName = "ToolsSerializers should build without throwing on a system that never loaded its reference.conf")]
public async Task Should_build_without_throwing_When_its_config_is_absent()
=> await ModuleSerializerSpecs.AssertBuildsWithoutModuleConfig("tools-no-config", s => new ToolsSerializers().Create(s));

[Fact(DisplayName = "Serialization should resolve reference.conf rows spelled as Akka.Hosting writes them when dynamic type loading is off")]
public async Task Should_resolve_assembly_qualified_names_When_dynamic_type_loading_is_disabled()
=> await ModuleSerializerSpecs.AssertHostingSpellingResolves("tools-aqn", ToolsRows, Sys);
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -6,8 +6,8 @@
//-----------------------------------------------------------------------

#nullable enable
using System;
using System.Collections.Generic;
using System.Collections.Immutable;
using Akka.Actor;
using Akka.Cluster.Tools.Client;
using Akka.Cluster.Tools.Client.Serialization;
using Akka.Cluster.Tools.PublishSubscribe;
Expand All @@ -25,21 +25,13 @@ namespace Akka.Cluster.Tools
/// </summary>
internal sealed class ToolsSerializers : ModuleSerializers
{
// the constructor reflection picks for each feature's reference.conf; each class has one constructor, so it always gets that one
public override IReadOnlyList<ModuleSerializer> Serializers { get; } = new[]
{
new ModuleSerializer(typeof(ClusterClientMessageSerializer), (system, _) => new ClusterClientMessageSerializer(system)),
new ModuleSerializer(typeof(DistributedPubSubMessageSerializer), (system, _) => new DistributedPubSubMessageSerializer(system)),
new ModuleSerializer(typeof(ClusterSingletonMessageSerializer), (system, _) => new ClusterSingletonMessageSerializer(system)),
};

public override IReadOnlyList<Type> BoundTypes { get; } = new[]
{
typeof(IClusterClientMessage),
typeof(IClusterClientProtocolMessage),
typeof(IDistributedPubSubMessage),
typeof(SendToOneSubscriber),
typeof(IClusterSingletonMessage),
};
public override ImmutableHashSet<SerializerDetails> Create(ExtendedActorSystem system) => ImmutableHashSet.Create(
SerializerDetails.Create("akka-cluster-client", new ClusterClientMessageSerializer(system),
ImmutableHashSet.Create(typeof(IClusterClientMessage), typeof(IClusterClientProtocolMessage))),
SerializerDetails.Create("akka-pubsub", new DistributedPubSubMessageSerializer(system),
ImmutableHashSet.Create(typeof(IDistributedPubSubMessage), typeof(SendToOneSubscriber))),
SerializerDetails.Create("akka-singleton", new ClusterSingletonMessageSerializer(system),
ImmutableHashSet.Create(typeof(IClusterSingletonMessage)))
);
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -6,8 +6,8 @@
//-----------------------------------------------------------------------

#nullable enable
using System.Linq;
using System.Threading.Tasks;
using Akka.Actor;
using Akka.Configuration;
using Akka.DistributedData.Serialization;
using Akka.Serialization;
Expand Down Expand Up @@ -43,9 +43,13 @@ public DistributedDataSerializersSpec(ITestOutputHelper output) : base(DDataRows
public void Should_have_a_reference_conf_row_When_the_table_lists_a_type()
{
var table = new DistributedDataSerializers();
ModuleSerializerSpecs.AssertTableMatchesConfig(DDataRows, table.Serializers.Select(s => s.Type), table.BoundTypes);
ModuleSerializerSpecs.AssertTableMatchesConfig(DDataRows, table.Create((ExtendedActorSystem)Sys));
}

[Fact(DisplayName = "DistributedDataSerializers should build without throwing on a system that never loaded its reference.conf")]
public async Task Should_build_without_throwing_When_its_config_is_absent()
=> await ModuleSerializerSpecs.AssertBuildsWithoutModuleConfig("ddata-no-config", s => new DistributedDataSerializers().Create(s));

[Fact(DisplayName = "Serialization should resolve reference.conf rows spelled as Akka.Hosting writes them when dynamic type loading is off")]
public async Task Should_resolve_assembly_qualified_names_When_dynamic_type_loading_is_disabled()
=> await ModuleSerializerSpecs.AssertHostingSpellingResolves("ddata-aqn", DDataRows, Sys);
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -6,8 +6,8 @@
//-----------------------------------------------------------------------

#nullable enable
using System;
using System.Collections.Generic;
using System.Collections.Immutable;
using Akka.Actor;
using Akka.Serialization;

namespace Akka.DistributedData.Serialization
Expand All @@ -17,17 +17,11 @@ namespace Akka.DistributedData.Serialization
/// </summary>
internal sealed class DistributedDataSerializers : ModuleSerializers
{
// the constructor reflection picks for reference.conf; each class has one constructor, so it always gets that one
public override IReadOnlyList<ModuleSerializer> Serializers { get; } = new[]
{
new ModuleSerializer(typeof(ReplicatedDataSerializer), (system, _) => new ReplicatedDataSerializer(system)),
new ModuleSerializer(typeof(ReplicatorMessageSerializer), (system, _) => new ReplicatorMessageSerializer(system)),
};

public override IReadOnlyList<Type> BoundTypes { get; } = new[]
{
typeof(IReplicatedDataSerialization),
typeof(IReplicatorMessage),
};
public override ImmutableHashSet<SerializerDetails> Create(ExtendedActorSystem system) => ImmutableHashSet.Create(
SerializerDetails.Create("akka-replicated-data", new ReplicatedDataSerializer(system),
ImmutableHashSet.Create(typeof(IReplicatedDataSerialization))),
SerializerDetails.Create("akka-data-replication", new ReplicatorMessageSerializer(system),
ImmutableHashSet.Create(typeof(IReplicatorMessage)))
);
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -157,7 +157,8 @@ private TVal Get(TKey key, int startIndex)
public ReplicatorMessageSerializer(Akka.Actor.ExtendedActorSystem system) : base(system)
{
_ser = new SerializationSupport(system);
var cacheTtl = system.Settings.Config.GetTimeSpan("akka.cluster.distributed-data.serializer-cache-time-to-live");
// reference.conf's default, for a system that never loaded it
var cacheTtl = system.Settings.Config.GetTimeSpan("akka.cluster.distributed-data.serializer-cache-time-to-live", TimeSpan.FromSeconds(10));
_readCache = new SmallCache<Read, byte[]>(4, cacheTtl, m => ReadToProto(m).ToByteArray());
_writeCache = new SmallCache<Write, byte[]>(4, cacheTtl, m => WriteToProto(m).ToByteArray());

Expand Down
20 changes: 10 additions & 10 deletions src/core/Akka.Cluster.Tests/Serialization/ClusterSerializersSpec.cs
Original file line number Diff line number Diff line change
Expand Up @@ -8,7 +8,6 @@
#nullable enable
using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading.Tasks;
using Akka.Actor;
using Akka.Cluster.Configuration;
Expand Down Expand Up @@ -76,22 +75,19 @@ private static AkkaSerialization Build(ActorSystem system, ModuleSerializerTable
public void Should_match_reflection_When_resolving_every_Cluster_conf_row()
{
var table = new ClusterSerializers();
ModuleSerializerSpecs.AssertTableMatchesConfig(ClusterRows, table.Serializers.Select(s => s.Type), table.BoundTypes);
var details = table.Create((ExtendedActorSystem)Sys);
ModuleSerializerSpecs.AssertTableMatchesConfig(ClusterRows, details);

// core's module map names Akka.Cluster, and ClusterSerializers is what it loads for it
ModuleSerializerTable.Default.ForAssembly("Akka.Cluster").Should().NotBeNull();

var reflected = Build(Sys, NoModules, dynamicTypeLoading: true);
var fromTable = Build(Sys, ModuleSerializerTable.Default, dynamicTypeLoading: true);
var settings = Sys.Settings.Config.GetConfig("akka.actor.serialization-settings");
var aliasByType = ModuleSerializerSpecs.SerializerRows(ClusterRows)
.ToDictionary(r => Type.GetType(r.TypeName, throwOnError: true)!, r => r.Alias);

foreach (var (type, create) in table.Serializers.Select(s => (s.Type, s.Create)))
foreach (var entry in details)
{
var built = create((ExtendedActorSystem)Sys, settings.GetConfig(aliasByType[type]));
built.Should().BeOfType(reflected.GetSerializerById(built.Identifier).GetType(), type.Name);
fromTable.GetSerializerById(built.Identifier).Should().BeOfType(type, type.Name);
entry.Serializer.Should().BeOfType(reflected.GetSerializerById(entry.Serializer.Identifier).GetType(), entry.Alias);
fromTable.GetSerializerById(entry.Serializer.Identifier).Should().BeOfType(entry.Serializer.GetType(), entry.Alias);
}

foreach (var type in BoundSamples)
Expand All @@ -102,9 +98,13 @@ public void Should_match_reflection_When_resolving_every_Cluster_conf_row()
public void Should_have_a_Cluster_conf_row_When_the_table_lists_a_type()
{
var table = new ClusterSerializers();
ModuleSerializerSpecs.AssertTableMatchesConfig(ClusterRows, table.Serializers.Select(s => s.Type), table.BoundTypes);
ModuleSerializerSpecs.AssertTableMatchesConfig(ClusterRows, table.Create((ExtendedActorSystem)Sys));
}

[Fact(DisplayName = "ClusterSerializers should build without throwing on a system that never loaded Cluster.conf")]
public async Task Should_build_without_throwing_When_Cluster_conf_is_absent()
=> await ModuleSerializerSpecs.AssertBuildsWithoutModuleConfig("cluster-no-config", s => new ClusterSerializers().Create(s));

[Fact(DisplayName = "Serialization should resolve Cluster.conf rows spelled as Akka.Hosting writes them when dynamic type loading is off")]
public async Task Should_resolve_assembly_qualified_names_When_dynamic_type_loading_is_disabled()
=> await ModuleSerializerSpecs.AssertHostingSpellingResolves("cluster-aqn", ClusterRows, Sys, BoundSamples);
Expand Down
23 changes: 8 additions & 15 deletions src/core/Akka.Cluster/Serialization/ClusterSerializers.cs
Original file line number Diff line number Diff line change
Expand Up @@ -6,8 +6,8 @@
//-----------------------------------------------------------------------

#nullable enable
using System;
using System.Collections.Generic;
using System.Collections.Immutable;
using Akka.Actor;
using Akka.Cluster.Routing;
using Akka.Delivery.Internal;
using Akka.Serialization;
Expand All @@ -19,18 +19,11 @@ namespace Akka.Cluster.Serialization
/// </summary>
internal sealed class ClusterSerializers : ModuleSerializers
{
// the constructor reflection picks for Cluster.conf; each class has one constructor, so it always gets that one
public override IReadOnlyList<ModuleSerializer> Serializers { get; } = new[]
{
new ModuleSerializer(typeof(ClusterMessageSerializer), (system, _) => new ClusterMessageSerializer(system)),
new ModuleSerializer(typeof(ReliableDeliverySerializer), (system, _) => new ReliableDeliverySerializer(system)),
};

public override IReadOnlyList<Type> BoundTypes { get; } = new[]
{
typeof(IClusterMessage),
typeof(ClusterRouterPool),
typeof(IDeliverySerializable),
};
public override ImmutableHashSet<SerializerDetails> Create(ExtendedActorSystem system) => ImmutableHashSet.Create(
SerializerDetails.Create("akka-cluster", new ClusterMessageSerializer(system),
ImmutableHashSet.Create(typeof(IClusterMessage), typeof(ClusterRouterPool))),
SerializerDetails.Create("reliable-delivery", new ReliableDeliverySerializer(system),
ImmutableHashSet.Create(typeof(IDeliverySerializable)))
);
}
}
Loading
Loading