diff --git a/RELEASE_NOTES.md b/RELEASE_NOTES.md index 1b446cc03ac..2b32b7bda2a 100644 --- a/RELEASE_NOTES.md +++ b/RELEASE_NOTES.md @@ -1,3 +1,22 @@ +#### 1.4.36 April 4 2022 #### +Akka.NET v1.4.36 is a minor release that contains some bug fixes. Most of the changes have been aimed at improving our web documentation and code cleanup to modernize some of our code. + +* [Akka: Bump Hyperion to 0.12.2](https://github.com/akkadotnet/akka.net/pull/5805) + +__Bug fixes__: +* [Akka: Fix CoordinatedShutdown memory leak](https://github.com/akkadotnet/akka.net/pull/5816) +* [Akka: Fix TcpConnection error handling and death pact de-registration](https://github.com/akkadotnet/akka.net/pull/5817) + +If you want to see the [full set of changes made in Akka.NET v1.4.36, click here](https://github.com/akkadotnet/akka.net/milestone/67?closed=1). + +| COMMITS | LOC+ | LOC- | AUTHOR | +|---------|------|------|---------------------| +| 5 | 274 | 33 | Gregorius Soedharmo | +| 4 | 371 | 6 | Ebere Abanonu | +| 3 | 9 | 3 | Aaron Stannard | +| 1 | 34 | 38 | Ismael Hamed | +| 1 | 2 | 3 | Adrian Leonhard | + #### 1.4.35 March 18 2022 #### Akka.NET v1.4.35 is a minor release that contains some bug fixes. Most of the changes have been aimed at improving our web documentation and code cleanup to modernize some of our code. diff --git a/build-system/pr-validation.yaml b/build-system/pr-validation.yaml index 315e664b808..5a1d268aa74 100644 --- a/build-system/pr-validation.yaml +++ b/build-system/pr-validation.yaml @@ -30,7 +30,7 @@ jobs: - task: Npm@1 inputs: command: "custom" - customCommand: "install -g cspell" + customCommand: "install -g cspell@5.17.0" - task: CmdLine@2 inputs: script: 'cspell --config ./docs/cSpell.json "docs/**/*.md"' diff --git a/docs/articles/clustering/cluster-overview.md b/docs/articles/clustering/cluster-overview.md index a022b6a1ecd..d2c40929a9c 100644 --- a/docs/articles/clustering/cluster-overview.md +++ b/docs/articles/clustering/cluster-overview.md @@ -236,6 +236,7 @@ A node might also exit the cluster gracefully, preventing it from being marked a +* [Cluster Member Roles](xref:member-roles) * [How to Create Scalable Clustered Akka.NET Apps Using Akka.Cluster](https://petabridge.com/blog/intro-to-akka-cluster/) * [Video: Introduction to Akka.Cluster](https://www.youtube.com/watch?v=mUTKvGyxbOA) * [Gossip Protocol](https://en.wikipedia.org/wiki/Gossip_protocol) diff --git a/docs/articles/clustering/member-roles.md b/docs/articles/clustering/member-roles.md new file mode 100644 index 00000000000..80bf89785ec --- /dev/null +++ b/docs/articles/clustering/member-roles.md @@ -0,0 +1,317 @@ +--- +uid: member-roles +title: Member Roles +--- + +# Member Roles + +![cluster roles](/images/cluster/cluster-roles.png) + +A cluster can have multiple nodes(machines/servers/vms) with different capabilities. +When you require an application to run on a node(machine/server/vm) with certain capabilities, roles helps you to distinguish the nodes so that application can be deployed on that node. +Specifying cluster role(s) is a best practice; you don't want an application that requires less computational power running and consuming resources meant for a mission-critical and resource-intensive application. +Even if you only have a single type of node in your cluster, you should still use roles for it so you can leverage this infrastructure as your cluster expands in the future; and they add zero overhead in any conceivable way. + +# How To Configure Cluster Roles + +Below I will show you how the cluster above can be reproduced. I will create a five-nodes(ActorSystems - all having same name, though, but living on different machine/server/vm) cluster with different roles applied: + +**Node1**: All of my code that receives requests from users and push same to the cluster will be deployed here! + +```hocon +akka +{ + cluster + { + roles = ["web"] + } +} +``` + +**Node2**: All of my code handling fraud detections will be deployed on this node + +```hocon +akka +{ + cluster + { + roles = ["fraud"] + } +} +``` + +**Node3**: All me code that retrieves, stores data will be deployed on this node + +```hocon +akka +{ + cluster + { + roles = ["storage"] + } +} +``` + +**Node4**: All my code that handles customer orders will be deployed on this node + +```hocon +akka +{ + cluster + { + roles = ["order"] + } +} +``` + +**Node5**: All my code that handles customer billing will be deployed on this node + +```hocon +akka +{ + cluster + { + roles = ["billing"] + } +} +``` + +Now that we have laid the foundation for what is to follow, Akka.Cluster is made of various ready-made extensions(or modules) you can deploy. +I will show you how you can deploy them on any of the nodes. Apart from the Akka.Cluster modules, if you just want to use the Akka.Cluster core, I will show you how you can deploy your own actor to the cluster node with the required role: + +**Cluster Sharding**: Sharding be will deployed on the nodes with the `order` role, `node4` + +```hocon +akka +{ + cluster + { + roles = ["order"] + sharding + { + role = "order" + } + } +} +``` + +**Distributed Pub-Sub**: DistributedPubSub will be deployed on the nodes with the `web` role, `node1`. + +```hocon +akka +{ + cluster + { + roles = ["web"] + pub-sub + { + role = "web" + } + } +} +``` + +**Distributed Data**: DistributedData will be deployed on the node with the `storage` role, `node3`. + +```hocon +akka +{ + cluster + { + roles = ["storage"] + distributed-data + { + role = "storage" + } + } +} +``` + +**Cluster Singleton**: To avoid over charging a customer more than once, my code will be deployed with `ClusterSingleton` on the node with the `billing` role, `node5` + +```hocon +akka +{ + cluster + { + roles = ["billing"] + singleton + { + role = "billing" + } + } +} +``` + +I have one more node, `node2`, with nothing running in it. I will deploy my custom fraud detection code there, and the way to do that is: + +```csharp +var selfMember = Cluster.Get(_actorSystem).SelfMember; +if (selfMember.HasRole("fraud")) +{ + context.ActorOf(Billing.Prop(), "bill-gate"); +} +else +{ + //sleep, probably! +} +``` + +Using the Cluster `SelfMember`, I am checking if the current node has the `billing` role and if yes, create the `Billing` actor. + +## Cluster-Aware Router + +Cluster-Aware routers automate how actors are deployed on the cluster and also how messages are routed based on the role specified! Routers in Akka.NET can be either grouped or pooled and you can read up on them [Routers](https://getakka.net/articles/actors/routers.html) + +**Router Group**: I will create Cluster-Aware Router Group for all my applications above! + +```hocon +akka +{ + actor + { + provider = "Akka.Cluster.ClusterActorRefProvider, Akka.Cluster" + deployment + { + /webdispatcher + { + router = consistent-hashing-group # routing strategy + routees.paths = ["/user/web"] # path of routee on each node + nr-of-instances = 3 # max number of total routees + cluster + { + enabled = on + use-role = "web" + } + } + /frauddispatcher + { + router = consistent-hashing-group # routing strategy + routees.paths = ["/user/fraud"] # path of routee on each node + nr-of-instances = 3 # max number of total routees + cluster + { + enabled = on + use-role = "fraud" + } + } + /billingdispatcher + { + router = consistent-hashing-group # routing strategy + routees.paths = ["/user/billing"] # path of routee on each node + nr-of-instances = 3 # max number of total routees + cluster + { + enabled = on + use-role = "billing" + } + } + /orderdispatcher + { + router = consistent-hashing-group # routing strategy + routees.paths = ["/user/order"] # path of routee on each node + nr-of-instances = 3 # max number of total routees + cluster + { + enabled = on + use-role = "order" + } + } + /storagedispatcher + { + router = consistent-hashing-group # routing strategy + routees.paths = ["/user/storage"] # path of routee on each node + nr-of-instances = 3 # max number of total routees + cluster + { + enabled = on + use-role = "storage" + } + } + } + } +} +``` + +```csharp +var web = system.ActorOf("web"); +var fraud = system.ActorOf("fraud"); +var order = system.ActorOf("order"); +var billing = system.ActorOf("billing"); +var storage = system.ActorOf("storage"); + +var webRouter = system.ActorOf(Props.Empty.WithRouter(FromConfig.Instance),"webdispatcher"); +var fraudRouter = system.ActorOf(Props.Empty.WithRouter(FromConfig.Instance),"frauddispatcher"); +var orderRouter = system.ActorOf(Props.Empty.WithRouter(FromConfig.Instance),"orderdispatcher"); +var billingRouter = system.ActorOf(Props.Empty.WithRouter(FromConfig.Instance),"billingispatcher"); +var storageRouter = system.ActorOf(Props.Empty.WithRouter(FromConfig.Instance),"storagedispatcher"); +``` + +**Router Pool**: I will create Cluster-Aware Router Pool for all my applications above! + +```hocon +akka +{ + actor + { + provider = "Akka.Cluster.ClusterActorRefProvider, Akka.Cluster" + deployment + { + /webdispatcher + { + router = round-robin-pool # routing strategy + max-nr-of-instances-per-node = 5 + cluster + { + enabled = on + use-role = "web" + } + } + /frauddispatcher + { + router = round-robin-pool # routing strategy + max-nr-of-instances-per-node = 5 + cluster + { + enabled = on + use-role = "fraud" + } + } + /billingdispatcher + { + router = round-robin-pool # routing strategy + max-nr-of-instances-per-node = 5 + cluster + { + enabled = on + use-role = "billing" + } + } + /orderdispatcher + { + router = round-robin-pool # routing strategy + max-nr-of-instances-per-node = 5 + cluster + { + enabled = on + use-role = "order" + } + } + /storagedispatcher + { + router = round-robin-pool # routing strategy + max-nr-of-instances-per-node = 5 + cluster + { + enabled = on + use-role = "storage" + } + } + } + } +} +``` + +```csharp +var routerProps = Props.Empty.WithRouter(FromConfig.Instance); +``` diff --git a/docs/articles/clustering/toc.yml b/docs/articles/clustering/toc.yml index 396a85065fe..fd54bdc1590 100644 --- a/docs/articles/clustering/toc.yml +++ b/docs/articles/clustering/toc.yml @@ -1,5 +1,7 @@ - name: Overview href: cluster-overview.md +- name: Member Roles + href: member-roles.md - name: Cluster Routing href: cluster-routing.md - name: Cluster Configuration diff --git a/docs/articles/examples.md b/docs/articles/examples.md new file mode 100644 index 00000000000..251e771e9be --- /dev/null +++ b/docs/articles/examples.md @@ -0,0 +1,40 @@ +--- +uid: akka-examples +title: Example Akka.NET Projects +--- + +A curated lists of projects that use various modules of Akka.NET + +## IoT + +* [Fire Alert AWS Kinesis Sample](https://github.com/petabridge/kinesis-sample): Akka.NET Streams + Cluster + AWS Kinesis Sample + +* [Home Temperature Monitor](https://github.com/grofab95/HomeTemperatureMonitor): Arduino temperature monitoring project built on **Akka.Cluster** + +## Machine Learning + +* [ML.NET ASP.NET Core Web API Akka.NET Sample](https://github.com/lqdev/MlNetAspNetAkkaSample): Sample application that uses Akka.NET Actors to scale predictions in an ASP.NET Core Web API. + +## Blockchain + +* [Blocksharp](https://github.com/revoltez/Blocksharp): A fully implemented Blockchain based on the Actor model using Akka.NET. + +## Web + +* [Cluster Web Crawler](https://github.com/petabridge/Cluster.WebCrawler): K8s, DevOps version of the Akka.Cluster WebCrawler code sample + +## Finance - Stock Trading + +* [In-Memory Stock Trading CQRS Sample](https://github.com/Aaronontheweb/InMemoryCQRSReplication): Akka.NET Reference Architecture - CQRS + Sharding + In-Memory Replication + +## Files + +* [File Processor](https://github.com/cgstevens/FileProcessor): The example demonstrates the use of Akka.NET Clustering to monitor the file system for new files. + +## Frameworks Built On Top Akka.NET + +* [Akkatecture](https://github.com/AfterLutz/Akkatecture): a cqrs and event sourcing framework for dotnet core using akka.NET. + +* [EventFly](https://github.com/Sporteco/EventFly): A cqrs and event sourcing framework for dotnet core, build on top of akka.net. Fully optimized around using akka's highly scalable message passing and event stream pub sub mechanisms. + +* [SharpPulsar](https://github.com/eaba/SharpPulsar): Apache Pulsar c# client built on top Akka.NET diff --git a/docs/articles/intro/getting-started/tutorial-1.md b/docs/articles/intro/getting-started/tutorial-1.md index e5f7bfb3857..ce899f4840a 100644 --- a/docs/articles/intro/getting-started/tutorial-1.md +++ b/docs/articles/intro/getting-started/tutorial-1.md @@ -45,7 +45,7 @@ _supervising_ every actor living as a child of them, i.e. under their path. We will explain supervision in more detail, all you need to know now is that every unhandled failure from actors bubbles up to their parent that, in turn, can decide how to handle this failure. These predefined actors are guardians in the -sense that they are the final lines of defence, where all unhandled failures +sense that they are the final lines of defense, where all unhandled failures from user, or system, actors end up. > Does the root guardian (the root path `/`) have a parent? As it turns out, it diff --git a/docs/articles/persistence/persistence-query.md b/docs/articles/persistence/persistence-query.md index 192842a8506..cb9759bd285 100644 --- a/docs/articles/persistence/persistence-query.md +++ b/docs/articles/persistence/persistence-query.md @@ -249,11 +249,11 @@ query ## Performance and Denormalization -When building systems using Event sourcing and CQRS ([Command & Query Responsibility Segregation](https://msdn.microsoft.com/en-us/library/jj554200.aspx)) techniques it is tremendously important to realise that the write-side has completely different needs from the read-side, and separating those concerns into datastores that are optimized for either side makes it possible to offer the best experience for the write and read sides independently. +When building systems using Event sourcing and CQRS ([Command & Query Responsibility Segregation](https://msdn.microsoft.com/en-us/library/jj554200.aspx)) techniques it is tremendously important to realize that the write-side has completely different needs from the read-side, and separating those concerns into datastores that are optimized for either side makes it possible to offer the best experience for the write and read sides independently. For example, in a bidding system it is important to "take the write" and respond to the bidder that we have accepted the bid as soon as possible, which means that write-throughput is of highest importance for the write-side – often this means that data stores which are able to scale to accommodate these requirements have a less expressive query side. -On the other hand the same application may have some complex statistics view or we may have analysts working with the data to figure out best bidding strategies and trends – this often requires some kind of expressive query capabilities like for example SQL or writing Spark jobs to analyse the data. Therefore the data stored in the write-side needs to be projected into the other read-optimized datastore. +On the other hand the same application may have some complex statistics view or we may have analysts working with the data to figure out best bidding strategies and trends – this often requires some kind of expressive query capabilities like for example SQL or writing Spark jobs to analyze the data. Therefore the data stored in the write-side needs to be projected into the other read-optimized datastore. > [!NOTE] > When referring to Materialized Views in Akka Persistence think of it as "some persistent storage of the result of a Query". In other words, it means that the view is created once, in order to be afterwards queried multiple times, as in this format it may be more efficient or interesting to query it (instead of the source events directly). diff --git a/docs/articles/serialization/serialization.md b/docs/articles/serialization/serialization.md index 464478a783e..b42e8d7656c 100644 --- a/docs/articles/serialization/serialization.md +++ b/docs/articles/serialization/serialization.md @@ -186,6 +186,41 @@ The only thing left to do for this class would be to fill in the serialization l Afterwards the configuration would need to be updated to reflect which name to bind to and the classes that use this serializer. +### Overriding Default Serializer Ids + +> [!WARNING] +> Changing the identification does not change the serialization binding, it only change the identification used by the serializer when it serialize a message. In fact, depending on the code, this might actually break the serializer as it might expect a specific identifier. You have been warned. + +Generally, overriding a default serializer identification is not recommended. The more recommended way is to change the serialization binding as was done when we [replace the default serializer with Hyperion](xref:serialization#how-to-setup-hyperion-as-the-default-serializer). + +In the rare case where you do need to override them, you can do it in one of two ways: + +* Overriding the `Identifier` property in your custom serializer class that inherits from the `Akka.Serialization.Serializer` abstract class. The `Identifier` property will override any HOCON settings; in fact, you will get a warning in your log when you do that, reminding you that you actually did that. +* Overriding the identifier inside your HOCON settings. To do this, you have to both change the original default serializer id and declare your own serializer using the original default serializer id. + +```c# + serialization-identifiers { + "Akka.Serialization.NewtonSoftJsonSerializer, Akka" : 1000001 + "MyAssembly.MyDefaultSerializer, SomeAssembly" : 1 +} + ``` + +this would in effect results this final HOCON settings: + +```c# +serialization-identifiers : { + "Akka.Serialization.ByteArraySerializer, Akka" : 4 + "Akka.Serialization.NewtonSoftJsonSerializer, Akka" : 1000001 + "Akka.Remote.Serialization.ProtobufSerializer, Akka.Remote" : 2 + "Akka.Remote.Serialization.DaemonMsgCreateSerializer, Akka.Remote" : 3 + "Akka.Remote.Serialization.MessageContainerSerializer, Akka.Remote" : 6 + "Akka.Remote.Serialization.MiscMessageSerializer, Akka.Remote" : 16 + "Akka.Remote.Serialization.PrimitiveSerializers, Akka.Remote" : 17 + "Akka.Remote.Serialization.SystemMessageSerializer, Akka.Remote" : 22 + "MyAssembly.MyDefaultSerializer, SomeAssembly" : 1 +} +``` + ### Programmatically Change NewtonSoft JSON Serializer Settings You can change the JSON serializer behavior by using the `NewtonSoftJsonSerializerSetup` class to programmatically @@ -354,7 +389,7 @@ from being deserialized: * `System.Diagnostics.Process` * `System.Management.IWbemClassObjectFreeThreaded` -Be warned that these class can be used as a man in the middle attack vector, but if you need +Be warned that these class can be used as a man in the middle and arbitrary code injection attack vector, but if you need to serialize one of these class, you can turn off this feature using this inside your HOCON settings: ```hocon diff --git a/docs/articles/toc.yml b/docs/articles/toc.yml index cfa782be1ff..adaa0a51ad7 100644 --- a/docs/articles/toc.yml +++ b/docs/articles/toc.yml @@ -23,4 +23,6 @@ - name: Configurations href: configuration/toc.yml - name: Serialization - href: serialization/toc.yml \ No newline at end of file + href: serialization/toc.yml +- name: Examples + href: examples.md \ No newline at end of file diff --git a/docs/cSpell.json b/docs/cSpell.json index f88553d8de5..562380cd646 100644 --- a/docs/cSpell.json +++ b/docs/cSpell.json @@ -105,7 +105,10 @@ "titlecase", "Varghese", "Vertech", - "DeadLetter" + "DeadLetter", + "Blocksharp", + "Akkatecture", + "SharpPulsar" ], "ignoreRegExpList": [ "\\((.*)\\)", // Markdown links diff --git a/docs/images/cluster/cluster-roles.png b/docs/images/cluster/cluster-roles.png new file mode 100644 index 00000000000..f448416b0fc Binary files /dev/null and b/docs/images/cluster/cluster-roles.png differ diff --git a/src/common.props b/src/common.props index 7e23057b9b7..204d9d08822 100644 --- a/src/common.props +++ b/src/common.props @@ -11,8 +11,9 @@ 2.4.1 + 2.4.3 17.0.0 - 0.12.0 + 0.12.2 [12.0.3,) 2.0.1 3.19.4 diff --git a/src/contrib/cluster/Akka.Cluster.Metrics.Tests.MultiNode/Akka.Cluster.Metrics.Tests.MultiNode.csproj b/src/contrib/cluster/Akka.Cluster.Metrics.Tests.MultiNode/Akka.Cluster.Metrics.Tests.MultiNode.csproj index d2ec0c471c0..70ff3acc750 100644 --- a/src/contrib/cluster/Akka.Cluster.Metrics.Tests.MultiNode/Akka.Cluster.Metrics.Tests.MultiNode.csproj +++ b/src/contrib/cluster/Akka.Cluster.Metrics.Tests.MultiNode/Akka.Cluster.Metrics.Tests.MultiNode.csproj @@ -16,6 +16,10 @@ + + all + runtime; build; native; contentfiles; analyzers; buildtransitive + diff --git a/src/contrib/cluster/Akka.Cluster.Sharding.Tests.MultiNode/Akka.Cluster.Sharding.Tests.MultiNode.csproj b/src/contrib/cluster/Akka.Cluster.Sharding.Tests.MultiNode/Akka.Cluster.Sharding.Tests.MultiNode.csproj index 68959b328df..0bbb5d93da2 100644 --- a/src/contrib/cluster/Akka.Cluster.Sharding.Tests.MultiNode/Akka.Cluster.Sharding.Tests.MultiNode.csproj +++ b/src/contrib/cluster/Akka.Cluster.Sharding.Tests.MultiNode/Akka.Cluster.Sharding.Tests.MultiNode.csproj @@ -15,9 +15,13 @@ - - + + + + all + runtime; build; native; contentfiles; analyzers; buildtransitive + diff --git a/src/contrib/cluster/Akka.Cluster.Tools.Tests.MultiNode/Akka.Cluster.Tools.Tests.MultiNode.csproj b/src/contrib/cluster/Akka.Cluster.Tools.Tests.MultiNode/Akka.Cluster.Tools.Tests.MultiNode.csproj index d0f9e768f29..26b4534b14c 100644 --- a/src/contrib/cluster/Akka.Cluster.Tools.Tests.MultiNode/Akka.Cluster.Tools.Tests.MultiNode.csproj +++ b/src/contrib/cluster/Akka.Cluster.Tools.Tests.MultiNode/Akka.Cluster.Tools.Tests.MultiNode.csproj @@ -14,9 +14,13 @@ - - + + + + all + runtime; build; native; contentfiles; analyzers; buildtransitive + diff --git a/src/contrib/cluster/Akka.DistributedData.Tests.MultiNode/Akka.DistributedData.Tests.MultiNode.csproj b/src/contrib/cluster/Akka.DistributedData.Tests.MultiNode/Akka.DistributedData.Tests.MultiNode.csproj index cfdb48fe75c..18f474d7123 100644 --- a/src/contrib/cluster/Akka.DistributedData.Tests.MultiNode/Akka.DistributedData.Tests.MultiNode.csproj +++ b/src/contrib/cluster/Akka.DistributedData.Tests.MultiNode/Akka.DistributedData.Tests.MultiNode.csproj @@ -14,9 +14,13 @@ - - + + + + all + runtime; build; native; contentfiles; analyzers; buildtransitive + diff --git a/src/contrib/dependencyinjection/Akka.DependencyInjection.Tests/BugFixSpec.cs b/src/contrib/dependencyinjection/Akka.DependencyInjection.Tests/BugFixSpec.cs new file mode 100644 index 00000000000..b08838bd1f5 --- /dev/null +++ b/src/contrib/dependencyinjection/Akka.DependencyInjection.Tests/BugFixSpec.cs @@ -0,0 +1,101 @@ +// //----------------------------------------------------------------------- +// // +// // Copyright (C) 2009-2022 Lightbend Inc. +// // Copyright (C) 2013-2022 .NET Foundation +// // +// //----------------------------------------------------------------------- + +using System; +using System.Threading; +using System.Threading.Tasks; +using Akka.Actor; +using Akka.Event; +using Akka.TestKit; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; +using Xunit; +using Xunit.Abstractions; +using FluentAssertions; +using static FluentAssertions.FluentActions; + +namespace Akka.DependencyInjection.Tests +{ + public class BugFixSpec: AkkaSpec, IAsyncLifetime + { + private readonly IServiceProvider _serviceProvider; + private readonly AkkaService _akkaService; + + public BugFixSpec(ITestOutputHelper output) : base(output) + { + var services = new ServiceCollection() + .AddSingleton() + .AddHostedService(); + + _serviceProvider = services.BuildServiceProvider(); + _akkaService = _serviceProvider.GetRequiredService(); + } + + [Fact(DisplayName = "DI should log an error if DI provider does not contain required parameter")] + public void ShouldLogAnErrorIfParameterInjectionFailed() + { + var system = _serviceProvider.GetRequiredService().ActorSystem; + var probe = CreateTestProbe(system); + system.EventStream.Subscribe(probe, typeof(Error)); + + var props = DependencyResolver.For(system).Props(); + var actor = system.ActorOf(props.WithDeploy(Deploy.Local), "testDIActor"); + + probe.ExpectMsg().Cause.Should().BeOfType(); + } + + internal class TestDiActor : ReceiveActor + { + public TestDiActor(NotInServices doesNotExistInDi) + { + } + } + + internal class NotInServices + { + } + + + public async Task InitializeAsync() + { + await _akkaService.StartAsync(default); + InitializeLogger(_akkaService.ActorSystem); + } + + public async Task DisposeAsync() + { + var sys = _serviceProvider.GetRequiredService().ActorSystem; + await sys.Terminate(); + } + + internal class AkkaService : IHostedService + { + public ActorSystem ActorSystem { get; private set; } + + private readonly IServiceProvider _serviceProvider; + + public AkkaService(IServiceProvider serviceProvider) + { + _serviceProvider = serviceProvider; + } + + public Task StartAsync(CancellationToken cancellationToken) + { + var setup = DependencyResolverSetup.Create(_serviceProvider) + .And(BootstrapSetup.Create().WithConfig(TestKitBase.DefaultConfig)); + + ActorSystem = ActorSystem.Create("TestSystem", setup); + return Task.CompletedTask; + } + + public async Task StopAsync(CancellationToken cancellationToken) + { + await ActorSystem.Terminate(); + } + } + } +} \ No newline at end of file diff --git a/src/contrib/serializers/Akka.Serialization.Hyperion.Tests/HyperionConfigTests.cs b/src/contrib/serializers/Akka.Serialization.Hyperion.Tests/HyperionConfigTests.cs index 33f8ccd1f84..8894886ef63 100644 --- a/src/contrib/serializers/Akka.Serialization.Hyperion.Tests/HyperionConfigTests.cs +++ b/src/contrib/serializers/Akka.Serialization.Hyperion.Tests/HyperionConfigTests.cs @@ -93,12 +93,14 @@ public void TypeFilter_defined_in_config_should_filter_serializer_properly(objec "); using (var system = ActorSystem.Create(nameof(HyperionConfigTests), config)) { - var serializer = (HyperionSerializer)system.Serialization.FindSerializerForType(typeof(object)); + var deserializer = (HyperionSerializer)system.Serialization.FindSerializerForType(typeof(object)); + var serializer = new HyperionSerializer(null, deserializer.Settings.WithDisallowUnsafeType(false)); - ((TypeFilter)serializer.Settings.TypeFilter).FilteredTypes.Count.Should().Be(2); + ((TypeFilter)deserializer.Settings.TypeFilter).FilteredTypes.Count.Should().Be(2); + var serialized = serializer.ToBinary(sampleObject); object deserialized = null; - Action act = () => deserialized = serializer.FromBinary(serialized); + Action act = () => deserialized = deserializer.FromBinary(serialized); if (shouldSucceed) { act.Should().NotThrow(); diff --git a/src/contrib/serializers/Akka.Serialization.Hyperion.Tests/HyperionSerializerSetupSpec.cs b/src/contrib/serializers/Akka.Serialization.Hyperion.Tests/HyperionSerializerSetupSpec.cs index a5e0cc8be2e..021623632d0 100644 --- a/src/contrib/serializers/Akka.Serialization.Hyperion.Tests/HyperionSerializerSetupSpec.cs +++ b/src/contrib/serializers/Akka.Serialization.Hyperion.Tests/HyperionSerializerSetupSpec.cs @@ -130,9 +130,10 @@ public void Setup_surrogate_should_work() [MemberData(nameof(DangerousObjectFactory))] public void Setup_disallow_unsafe_type_should_work(object dangerousObject, Type type) { - var serializer = new HyperionSerializer((ExtendedActorSystem)Sys, HyperionSerializerSettings.Default); + var deserializer = new HyperionSerializer((ExtendedActorSystem)Sys, HyperionSerializerSettings.Default); + var serializer = new HyperionSerializer((ExtendedActorSystem)Sys, deserializer.Settings.WithDisallowUnsafeType(false)); var serialized = serializer.ToBinary(dangerousObject); - serializer.Invoking(s => s.FromBinary(serialized, type)).Should().Throw(); + deserializer.Invoking(s => s.FromBinary(serialized, type)).Should().Throw(); } [Theory] @@ -146,12 +147,13 @@ public void Setup_TypeFilter_should_filter_types_properly(object sampleObject, b .Build()); var settings = setup.ApplySettings(HyperionSerializerSettings.Default); - var serializer = new HyperionSerializer((ExtendedActorSystem)Sys, settings); - - ((TypeFilter)serializer.Settings.TypeFilter).FilteredTypes.Count.Should().Be(2); + var deserializer = new HyperionSerializer((ExtendedActorSystem)Sys, settings); + var serializer = new HyperionSerializer((ExtendedActorSystem)Sys, deserializer.Settings.WithDisallowUnsafeType(false)); var serialized = serializer.ToBinary(sampleObject); + + ((TypeFilter)deserializer.Settings.TypeFilter).FilteredTypes.Count.Should().Be(2); object deserialized = null; - Action act = () => deserialized = serializer.FromBinary(serialized); + Action act = () => deserialized = deserializer.FromBinary(serialized); if (shouldSucceed) { act.Should().NotThrow(); diff --git a/src/contrib/serializers/Akka.Serialization.Hyperion/HyperionSerializer.cs b/src/contrib/serializers/Akka.Serialization.Hyperion/HyperionSerializer.cs index b05f0cc3b50..dc525668441 100644 --- a/src/contrib/serializers/Akka.Serialization.Hyperion/HyperionSerializer.cs +++ b/src/contrib/serializers/Akka.Serialization.Hyperion/HyperionSerializer.cs @@ -401,5 +401,75 @@ public HyperionSerializerSettings( DisallowUnsafeType = disallowUnsafeType; TypeFilter = typeFilter; } + + public HyperionSerializerSettings WithPreserveObjectReference(bool preserveObjectReferences) + => new HyperionSerializerSettings( + preserveObjectReferences: preserveObjectReferences, + versionTolerance: VersionTolerance, + knownTypesProvider: KnownTypesProvider, + packageNameOverrides: PackageNameOverrides, + surrogates: Surrogates, + disallowUnsafeType: DisallowUnsafeType, + typeFilter: TypeFilter); + + public HyperionSerializerSettings WithVersionTolerance(bool versionTolerance) + => new HyperionSerializerSettings( + preserveObjectReferences: PreserveObjectReferences, + versionTolerance: versionTolerance, + knownTypesProvider: KnownTypesProvider, + packageNameOverrides: PackageNameOverrides, + surrogates: Surrogates, + disallowUnsafeType: DisallowUnsafeType, + typeFilter: TypeFilter); + + public HyperionSerializerSettings WithKnownTypesProvider(Type knownTypesProvider) + => new HyperionSerializerSettings( + preserveObjectReferences: PreserveObjectReferences, + versionTolerance: VersionTolerance, + knownTypesProvider: knownTypesProvider, + packageNameOverrides: PackageNameOverrides, + surrogates: Surrogates, + disallowUnsafeType: DisallowUnsafeType, + typeFilter: TypeFilter); + + public HyperionSerializerSettings WithPackageNameOverrides(IEnumerable> packageNameOverrides) + => new HyperionSerializerSettings( + preserveObjectReferences: PreserveObjectReferences, + versionTolerance: VersionTolerance, + knownTypesProvider: KnownTypesProvider, + packageNameOverrides: packageNameOverrides, + surrogates: Surrogates, + disallowUnsafeType: DisallowUnsafeType, + typeFilter: TypeFilter); + + public HyperionSerializerSettings WithSurrogates(IEnumerable surrogates) + => new HyperionSerializerSettings( + preserveObjectReferences: PreserveObjectReferences, + versionTolerance: VersionTolerance, + knownTypesProvider: KnownTypesProvider, + packageNameOverrides: PackageNameOverrides, + surrogates: surrogates, + disallowUnsafeType: DisallowUnsafeType, + typeFilter: TypeFilter); + + public HyperionSerializerSettings WithDisallowUnsafeType(bool disallowUnsafeType) + => new HyperionSerializerSettings( + preserveObjectReferences: PreserveObjectReferences, + versionTolerance: VersionTolerance, + knownTypesProvider: KnownTypesProvider, + packageNameOverrides: PackageNameOverrides, + surrogates: Surrogates, + disallowUnsafeType: disallowUnsafeType, + typeFilter: TypeFilter); + + public HyperionSerializerSettings WithTypeFilter(ITypeFilter typeFilter) + => new HyperionSerializerSettings( + preserveObjectReferences: PreserveObjectReferences, + versionTolerance: VersionTolerance, + knownTypesProvider: KnownTypesProvider, + packageNameOverrides: PackageNameOverrides, + surrogates: Surrogates, + disallowUnsafeType: DisallowUnsafeType, + typeFilter: typeFilter); } } diff --git a/src/contrib/testkits/Akka.TestKit.Xunit2/Internals/Loggers.cs b/src/contrib/testkits/Akka.TestKit.Xunit2/Internals/Loggers.cs index cfefe9758de..accd94f94f4 100644 --- a/src/contrib/testkits/Akka.TestKit.Xunit2/Internals/Loggers.cs +++ b/src/contrib/testkits/Akka.TestKit.Xunit2/Internals/Loggers.cs @@ -8,6 +8,7 @@ using System; using Akka.Actor; using Akka.Event; +using Akka.Util; using Xunit.Abstractions; namespace Akka.TestKit.Xunit2.Internals @@ -58,7 +59,8 @@ private void HandleLogEvent(LogEvent e) } catch (InvalidOperationException ie) { - Console.WriteLine($"Received InvalidOperationException: {ie} - probably because the test had completed executing."); + StandardOutWriter.WriteLine($"Received InvalidOperationException: {ie} - probably because the test had completed executing."); + Context.Stop(Self); // shut ourselves down, can't do our job any longer } } } diff --git a/src/core/Akka.Cluster.Tests.MultiNode/Akka.Cluster.Tests.MultiNode.csproj b/src/core/Akka.Cluster.Tests.MultiNode/Akka.Cluster.Tests.MultiNode.csproj index 38652c8d6cc..88107065bcd 100644 --- a/src/core/Akka.Cluster.Tests.MultiNode/Akka.Cluster.Tests.MultiNode.csproj +++ b/src/core/Akka.Cluster.Tests.MultiNode/Akka.Cluster.Tests.MultiNode.csproj @@ -14,9 +14,13 @@ - - + + + + all + runtime; build; native; contentfiles; analyzers; buildtransitive + diff --git a/src/core/Akka.Cluster/ClusterDaemon.cs b/src/core/Akka.Cluster/ClusterDaemon.cs index 338b4a97469..188bfe4f0b2 100644 --- a/src/core/Akka.Cluster/ClusterDaemon.cs +++ b/src/core/Akka.Cluster/ClusterDaemon.cs @@ -2243,7 +2243,7 @@ private void ShutdownSelfWhenDown() { // the reason for not shutting down immediately is to give the gossip a chance to spread // the downing information to other downed nodes, so that they can shutdown themselves - _cluster.LogInfo("Shutting down myself"); + _cluster.LogInfo("Node has been marked as DOWN. Shutting down myself"); // not crucial to send gossip, but may speedup removal since fallback to failure detection is not needed // if other downed know that this node has seen the version SendGossipRandom(MaxGossipsBeforeShuttingDownMyself); diff --git a/src/core/Akka.Docs.Tests/Persistence/PersistentActor/PersistAsync.cs b/src/core/Akka.Docs.Tests/Persistence/PersistentActor/PersistAsync.cs index df5d6340ba6..9654dc43525 100644 --- a/src/core/Akka.Docs.Tests/Persistence/PersistentActor/PersistAsync.cs +++ b/src/core/Akka.Docs.Tests/Persistence/PersistentActor/PersistAsync.cs @@ -27,9 +27,8 @@ protected override void OnCommand(object message) if (message is string c) { Sender.Tell(c); - Persist($"evt-{c}-1", e => Sender.Tell(e)); - Persist($"evt-{c}-2", e => Sender.Tell(e)); - DeferAsync($"evt-{c}-3", e => Sender.Tell(e)); + PersistAsync($"evt-{c}-1", e => Sender.Tell(e)); + PersistAsync($"evt-{c}-2", e => Sender.Tell(e)); } } } diff --git a/src/core/Akka.Remote.Tests.MultiNode/Akka.Remote.Tests.MultiNode.csproj b/src/core/Akka.Remote.Tests.MultiNode/Akka.Remote.Tests.MultiNode.csproj index 14b3c52a0ff..917ee38dca9 100644 --- a/src/core/Akka.Remote.Tests.MultiNode/Akka.Remote.Tests.MultiNode.csproj +++ b/src/core/Akka.Remote.Tests.MultiNode/Akka.Remote.Tests.MultiNode.csproj @@ -15,6 +15,10 @@ + + all + runtime; build; native; contentfiles; analyzers; buildtransitive + diff --git a/src/core/Akka/Actor/ActorSystem.cs b/src/core/Akka/Actor/ActorSystem.cs index 318e0ac6846..6b1524c55ea 100644 --- a/src/core/Akka/Actor/ActorSystem.cs +++ b/src/core/Akka/Actor/ActorSystem.cs @@ -364,7 +364,7 @@ private static ActorSystem CreateAndStartSystem(string name, Config withFallback /// public abstract Task Terminate(); - internal abstract void FinalTerminate(); + internal abstract Task FinalTerminate(); /// /// Returns a task which will be completed after the has been diff --git a/src/core/Akka/Actor/CoordinatedShutdown.cs b/src/core/Akka/Actor/CoordinatedShutdown.cs index f1b78cad7ae..99c456f6ad2 100644 --- a/src/core/Akka/Actor/CoordinatedShutdown.cs +++ b/src/core/Akka/Actor/CoordinatedShutdown.cs @@ -270,7 +270,7 @@ private ClusterJoinUnsuccessfulReason() { } /// internal readonly List OrderedPhases; - private readonly ConcurrentBag>> _clrShutdownTasks = new ConcurrentBag>>(); + private readonly ConcurrentSet>> _clrShutdownTasks = new ConcurrentSet>>(); private readonly ConcurrentDictionary>)>> _tasks = new ConcurrentDictionary>)>>(); private readonly AtomicReference _runStarted = new AtomicReference(null); private readonly AtomicBoolean _clrHooksStarted = new AtomicBoolean(false); @@ -335,7 +335,7 @@ internal void AddClrShutdownHook(Func> hook) { if (!_clrHooksStarted) { - _clrShutdownTasks.Add(hook); + _clrShutdownTasks.TryAdd(hook); } } @@ -653,13 +653,13 @@ internal static void InitPhaseActorSystemTerminate(ActorSystem system, Config co if (terminateActorSystem) { - system.FinalTerminate(); - return system.Terminate().ContinueWith(tr => + return system.FinalTerminate().ContinueWith(tr => { if (exitClr && !coord._runningClrHook) { Environment.Exit(0); } + return Done.Instance; }); } @@ -691,7 +691,10 @@ internal static void InitClrHook(ActorSystem system, Config conf, CoordinatedShu var exitTask = TerminateOnClrExit(coord); // run all hooks during termination sequence AppDomain.CurrentDomain.ProcessExit += exitTask; - system.WhenTerminated.ContinueWith(tr => { AppDomain.CurrentDomain.ProcessExit -= exitTask; }); + system.WhenTerminated.ContinueWith(tr => + { + AppDomain.CurrentDomain.ProcessExit -= exitTask; + }); coord.AddClrShutdownHook(() => { diff --git a/src/core/Akka/Actor/Internal/ActorSystemImpl.cs b/src/core/Akka/Actor/Internal/ActorSystemImpl.cs index f7540c33f04..ed08b4acefa 100644 --- a/src/core/Akka/Actor/Internal/ActorSystemImpl.cs +++ b/src/core/Akka/Actor/Internal/ActorSystemImpl.cs @@ -525,20 +525,19 @@ public override Task Terminate() { if(Settings.CoordinatedShutdownRunByActorSystemTerminate) { - CoordinatedShutdown.Get(this).Run(CoordinatedShutdown.ActorSystemTerminateReason.Instance); - } else - { - FinalTerminate(); + return CoordinatedShutdown.Get(this) + .Run(CoordinatedShutdown.ActorSystemTerminateReason.Instance); } - return WhenTerminated; + return FinalTerminate(); } - internal override void FinalTerminate() + internal override Task FinalTerminate() { Log.Debug("System shutdown initiated"); if (!Settings.LogDeadLettersDuringShutdown && _logDeadLetterListener != null) Stop(_logDeadLetterListener); _provider.Guardian.Stop(); + return WhenTerminated; } /// diff --git a/src/core/Akka/Event/EventBusUnsubscriber.cs b/src/core/Akka/Event/EventBusUnsubscriber.cs index bf865c879bd..f4724a72c2f 100644 --- a/src/core/Akka/Event/EventBusUnsubscriber.cs +++ b/src/core/Akka/Event/EventBusUnsubscriber.cs @@ -90,7 +90,7 @@ protected override void PreStart() /// /// Registers a new subscriber to be death-watched and automatically unsubscribed. /// - internal class Register + internal class Register : INoSerializationVerificationNeeded { public Register(IActorRef actor) { @@ -108,7 +108,7 @@ public Register(IActorRef actor) /// /// Unsubscribes an actor that is no longer subscribed and does not need to be death-watched any longer. /// - internal class UnregisterIfNoMoreSubscribedChannels + internal class UnregisterIfNoMoreSubscribedChannels : INoSerializationVerificationNeeded { public UnregisterIfNoMoreSubscribedChannels(IActorRef actor) { diff --git a/src/core/Akka/IO/TcpConnection.cs b/src/core/Akka/IO/TcpConnection.cs index 5d87c6f6114..a25a5ce799c 100644 --- a/src/core/Akka/IO/TcpConnection.cs +++ b/src/core/Akka/IO/TcpConnection.cs @@ -13,7 +13,6 @@ using System.Linq; using System.Net.Sockets; using System.Runtime.CompilerServices; -using System.Threading; using Akka.Actor; using Akka.Dispatch; using Akka.Event; @@ -108,14 +107,14 @@ enum ConnectionStatus protected TcpConnection(TcpExt tcp, Socket socket, bool pullMode, Option writeCommandsBufferMaxSize) { if (socket == null) throw new ArgumentNullException(nameof(socket)); - + _pullMode = pullMode; _writeCommandsQueue = new PendingSimpleWritesQueue(Log, writeCommandsBufferMaxSize); _traceLogging = tcp.Settings.TraceLogging; - + Tcp = tcp; Socket = socket; - + if (pullMode) SetStatus(ConnectionStatus.ReadingSuspended); } @@ -155,30 +154,26 @@ private Receive WaitingForRegistration(IActorRef commander) // up to this point we've been watching the commander, // but since registration is now complete we only need to watch the handler from here on if (!Equals(register.Handler, commander)) - { - Context.Unwatch(commander); - Context.Watch(register.Handler); - } + SignDeathPact(register.Handler); // will unsign death pact with commander automatically if (_traceLogging) Log.Debug("[{0}] registered as connection handler", register.Handler); var registerInfo = new ConnectionInfo(register.Handler, register.KeepOpenOnPeerClosed, register.UseResumeWriting); - // if we have resumed reading from pullMode while waiting for Register then read - if (_pullMode && !HasStatus(ConnectionStatus.ReadingSuspended)) ResumeReading(); - else if (!_pullMode) ReceiveAsync(); - Context.SetReceiveTimeout(null); Context.Become(Connected(registerInfo)); + // if we are in push mode or already have resumed reading in pullMode while waiting for Register then read + if (!_pullMode || !HasStatus(ConnectionStatus.ReadingSuspended)) ResumeReading(); + // If there is something buffered before we got Register message - put it all to the socket var bufferedWrite = GetNextWrite(); if (bufferedWrite.HasValue) { SetStatus(ConnectionStatus.Sending); DoWrite(registerInfo, bufferedWrite.Value); - } - + } + return true; case ResumeReading _: ClearStatus(ConnectionStatus.ReadingSuspended); return true; case SuspendReading _: SetStatus(ConnectionStatus.ReadingSuspended); return true; @@ -205,7 +200,7 @@ private Receive WaitingForRegistration(IActorRef commander) Log.Warning("Received Write command before Register command. " + "It will be buffered until Register will be received (buffered write size is {0} bytes)", commandSize); } - + return true; default: return false; } @@ -268,7 +263,7 @@ private Receive ClosingWithPendingWrite(ConnectionInfo info, IActorRef closeComm AcknowledgeSent(); if (IsWritePending) DoWrite(info, GetAllowedPendingWrite()); - else + else HandleClose(info, closeCommander, closedEvent); return true; case UpdatePendingWriteAndThen updatePendingWrite: @@ -277,7 +272,7 @@ private Receive ClosingWithPendingWrite(ConnectionInfo info, IActorRef closeComm if (nextWrite.HasValue) DoWrite(info, nextWrite); - else + else HandleClose(info, closeCommander, closedEvent); return true; case WriteFileFailed fail: HandleError(info.Handler, fail.Cause); return true; @@ -312,7 +307,7 @@ private Receive HandleWriteMessages(ConnectionInfo info) case SocketSent _: // Send ack to sender AcknowledgeSent(); - + // If there is something to send - send it var pendingWrite = GetAllowedPendingWrite(); if (pendingWrite.HasValue) @@ -320,14 +315,14 @@ private Receive HandleWriteMessages(ConnectionInfo info) SetStatus(ConnectionStatus.Sending); DoWrite(info, pendingWrite); } - + // If message is fully sent, notify sender who sent ResumeWriting command if (!IsWritePending && _interestedInResume != null) { _interestedInResume.Tell(WritingResumed.Instance); _interestedInResume = null; } - + return true; case WriteCommand write: if (HasStatus(ConnectionStatus.WritingSuspended)) @@ -356,7 +351,7 @@ private Receive HandleWriteMessages(ConnectionInfo info) DropWrite(info, write); return true; } - + nextWrite = GetNextWrite(headCommands: new []{ (simpleWriteCommand, Sender) }); } else @@ -364,15 +359,15 @@ private Receive HandleWriteMessages(ConnectionInfo info) _writeCommandsQueue.EnqueueSimpleWrites(write, Sender); nextWrite = GetNextWrite(); } - + // If there is something to send and we are allowed to, lets put the next command on the wire if (nextWrite.HasValue) { SetStatus(ConnectionStatus.Sending); DoWrite(info, nextWrite.Value); - } + } } - + return true; case ResumeWriting _: /* @@ -396,7 +391,7 @@ private Receive HandleWriteMessages(ConnectionInfo info) case UpdatePendingWriteAndThen updatePendingWrite: var updatedWrite = updatePendingWrite.RemainingWrite; updatePendingWrite.Work(); - if (updatedWrite.HasValue) + if (updatedWrite.HasValue) DoWrite(info, updatedWrite.Value); return true; case WriteFileFailed fail: @@ -461,7 +456,7 @@ private void AcknowledgeSent() { ackInfo.Commander.Tell(ackInfo.Ack); } - + ClearStatus(ConnectionStatus.Sending); } @@ -538,7 +533,7 @@ private void DoWrite(ConnectionInfo info, Option write) { _pendingAcks.Enqueue(pendingAck); } - + write.Value.DoWrite(info); } @@ -672,7 +667,7 @@ SocketCompleted ResolveMessage(SocketAsyncEventArgs e) throw new NotSupportedException($"Socket operation {e.LastOperation} is not supported"); } } - + var args = new SocketAsyncEventArgs(); args.UserToken = onCompleteNotificationsReceiver; args.Completed += (sender, e) => @@ -684,7 +679,7 @@ SocketCompleted ResolveMessage(SocketAsyncEventArgs e) return args; } - + protected void ReleaseSocketEventArgs(SocketAsyncEventArgs e) { e.UserToken = null; @@ -700,7 +695,7 @@ protected void ReleaseSocketEventArgs(SocketAsyncEventArgs e) catch (InvalidOperationException) { } e.Dispose(); - + } [MethodImpl(MethodImplOptions.AggressiveInlining)] @@ -728,6 +723,7 @@ private void Abort() protected void StopWith(CloseInformation closeInfo) { _closedMessage = closeInfo; + UnsignDeathPact(); Context.Stop(Self); } @@ -810,7 +806,7 @@ private Option GetNextWrite(IEnumerable<(SimpleWriteCommand Comman { return CreatePendingBufferWrite(writeCommands); } - + // No more writes out there return Option.None; } @@ -986,7 +982,7 @@ public bool EnqueueSimpleWrites(WriteCommand command, IActorRef sender) { return EnqueueSimpleWrites(command, sender, out _); } - + /// /// Adds all subcommands stored in provided command. /// Performs buffer size checks @@ -997,7 +993,7 @@ public bool EnqueueSimpleWrites(WriteCommand command, IActorRef sender) public bool EnqueueSimpleWrites(WriteCommand command, IActorRef sender, out int bufferedSize) { bufferedSize = 0; - + foreach (var writeInfo in ExtractFromCommand(command)) { var sizeAfterAppending = _totalSizeInBytes + writeInfo.DataSize; @@ -1013,10 +1009,10 @@ public bool EnqueueSimpleWrites(WriteCommand command, IActorRef sender, out int _queue.Enqueue((writeInfo.Command, sender, writeInfo.DataSize)); bufferedSize += writeInfo.DataSize; } - + return true; } - + /// /// Adds all subcommands stored in provided command. /// Performs buffer size checks for all, except first one, that is not buffered @@ -1037,7 +1033,7 @@ public bool EnqueueSimpleWritesExceptFirst(WriteCommand command, IActorRef sende first = writeInfo.Command; continue; } - + var sizeAfterAppending = _totalSizeInBytes + writeInfo.DataSize; if (_maxQueueSizeInBytes.HasValue && _maxQueueSizeInBytes.Value < sizeAfterAppending) { @@ -1061,7 +1057,7 @@ public bool EnqueueSimpleWritesExceptFirst(WriteCommand command, IActorRef sende { if (_queue.Count == 0) throw new InvalidOperationException("Write commands queue is empty"); - + var (command, sender, size) = _queue.Dequeue(); _totalSizeInBytes -= size; return (command, sender); @@ -1076,7 +1072,7 @@ public bool EnqueueSimpleWritesExceptFirst(WriteCommand command, IActorRef sende while (TryGetNext(out var command)) yield return command; } - + /// /// Gets next command from the queue, if any ///