Akka.IO: redesign TcpConnection

src/core/Akka.API.Tests/verify/CoreAPISpec.ApproveCore.DotNet.verified.txt

@@ -2269,6 +2269,7 @@ namespace Akka.Configuration
         public virtual System.Collections.Generic.IList<bool> GetBooleanList(string path) { }
         public virtual System.Collections.Generic.IList<byte> GetByteList(string path) { }
         public virtual System.Nullable<long> GetByteSize(string path) { }
+        [return: System.Diagnostics.CodeAnalysis.NotNullIfNotNullAttribute("def")]
         public virtual System.Nullable<long> GetByteSize(string path, System.Nullable<long> def = null) { }
         public virtual Akka.Configuration.Config GetConfig(string path) { }
         public virtual decimal GetDecimal(string path, [System.Runtime.CompilerServices.DecimalConstantAttribute(0, 0, 0u, 0u, 0u)] decimal @default) { }
@@ -4001,6 +4002,8 @@ namespace Akka.IO
             public System.Net.EndPoint LocalAddress { get; }
             public System.Collections.Generic.IEnumerable<Akka.IO.Inet.SocketOption> Options { get; }
             public bool PullMode { get; }
+            [System.Runtime.CompilerServices.NullableAttribute(2)]
+            public Akka.IO.TcpSettings TcpSettings { get; set; }
             public override string ToString() { }
         }
         public sealed class Bound : Akka.IO.Tcp.Event
@@ -4043,6 +4046,7 @@ namespace Akka.IO
         public sealed class CompoundWrite : Akka.IO.Tcp.WriteCommand, System.Collections.Generic.IEnumerable<Akka.IO.Tcp.SimpleWriteCommand>, System.Collections.IEnumerable
         {
             public CompoundWrite(Akka.IO.Tcp.SimpleWriteCommand head, Akka.IO.Tcp.WriteCommand tailCommand) { }
+            public override long Bytes { get; }
             public Akka.IO.Tcp.SimpleWriteCommand Head { get; }
             public Akka.IO.Tcp.WriteCommand TailCommand { get; }
             public System.Collections.Generic.IEnumerator<Akka.IO.Tcp.SimpleWriteCommand> GetEnumerator() { }
@@ -4065,6 +4069,8 @@ namespace Akka.IO
             public System.Collections.Generic.IEnumerable<Akka.IO.Inet.SocketOption> Options { get; }
             public bool PullMode { get; }
             public System.Net.EndPoint RemoteAddress { get; }
+            [System.Runtime.CompilerServices.NullableAttribute(2)]
+            public Akka.IO.TcpSettings TcpSettings { get; set; }
             public System.Nullable<System.TimeSpan> Timeout { get; }
             public override string ToString() { }
         }
@@ -4184,6 +4190,7 @@ namespace Akka.IO
         {
             public static readonly Akka.IO.Tcp.Write Empty;
             public override Akka.IO.Tcp.Event Ack { get; }
+            public override long Bytes { get; }
             public Akka.IO.ByteString Data { get; }
             public static Akka.IO.Tcp.Write Create(Akka.IO.ByteString data) { }
             public static Akka.IO.Tcp.Write Create(Akka.IO.ByteString data, Akka.IO.Tcp.Event ack) { }
@@ -4192,6 +4199,7 @@ namespace Akka.IO
         public abstract class WriteCommand : Akka.IO.Tcp.Command
         {
             protected WriteCommand() { }
+            public abstract long Bytes { get; }
             public static Akka.IO.Tcp.WriteCommand Create(System.Collections.Generic.IEnumerable<Akka.IO.Tcp.WriteCommand> writes) { }
             public static Akka.IO.Tcp.WriteCommand Create(params WriteCommand[] writes) { }
             public Akka.IO.Tcp.CompoundWrite Prepend(Akka.IO.Tcp.SimpleWriteCommand other) { }
@@ -4205,7 +4213,6 @@ namespace Akka.IO
     public sealed class TcpExt : Akka.IO.IOExtension
     {
         public TcpExt(Akka.Actor.ExtendedActorSystem system) { }
-        public Akka.IO.Buffers.IBufferPool BufferPool { get; }
         public override Akka.Actor.IActorRef Manager { get; }
         public Akka.IO.TcpSettings Settings { get; }
     }
@@ -4231,21 +4238,31 @@ namespace Akka.IO
         public static Akka.IO.Tcp.Command Unbind() { }
         public static Akka.IO.Tcp.Command Write(Akka.IO.ByteString data, Akka.IO.Tcp.Event ack = null) { }
     }
-    public class TcpSettings
+    public sealed class TcpSettings : System.IEquatable<Akka.IO.TcpSettings>
     {
+        [System.ObsoleteAttribute("Many of these options are no longer used. Use the TcpSettings.Create method inste" +
+            "ad.")]
         public TcpSettings(string bufferPoolConfigPath, int initialSocketAsyncEventArgs, bool traceLogging, int batchAcceptLimit, System.Nullable<System.TimeSpan> registerTimeout, int receivedMessageSizeLimit, string managementDispatcher, string fileIoDispatcher, int transferToLimit, int finishConnectRetries, bool outgoingSocketForceIpv4, int writeCommandsQueueMaxSize) { }
-        public int BatchAcceptLimit { get; }
+        public int BatchAcceptLimit { get; set; }
+        [System.ObsoleteAttribute("This property is unused")]
         public string BufferPoolConfigPath { get; }
+        [System.ObsoleteAttribute("This property is unused")]
         public string FileIODispatcher { get; }
-        public int FinishConnectRetries { get; }
+        public int FinishConnectRetries { get; set; }
+        [System.ObsoleteAttribute("This property is unused")]
         public int InitialSocketAsyncEventArgs { get; }
         public string ManagementDispatcher { get; }
-        public bool OutgoingSocketForceIpv4 { get; }
-        public int ReceivedMessageSizeLimit { get; }
-        public System.Nullable<System.TimeSpan> RegisterTimeout { get; }
-        public bool TraceLogging { get; }
+        public int MaxFrameSizeBytes { get; set; }
+        public bool OutgoingSocketForceIpv4 { get; set; }
+        public int ReceiveBufferSize { get; set; }
+        [System.ObsoleteAttribute("This property is now MaxFrameSizeBytes")]
+        public long ReceivedMessageSizeLimit { get; }
+        public System.Nullable<System.TimeSpan> RegisterTimeout { get; set; }
+        public int SendBufferSize { get; set; }
+        public bool TraceLogging { get; set; }
+        [System.ObsoleteAttribute("This property is unused")]
         public int TransferToLimit { get; set; }
-        public int WriteCommandsQueueMaxSize { get; }
+        public int WriteCommandsQueueMaxSize { get; set; }
         public static Akka.IO.TcpSettings Create(Akka.Actor.ActorSystem system) { }
         public static Akka.IO.TcpSettings Create(Akka.Configuration.Config config) { }
     }

src/core/Akka.API.Tests/verify/CoreAPISpec.ApproveCore.Net.verified.txt

@@ -3991,6 +3991,8 @@ namespace Akka.IO
             public System.Net.EndPoint LocalAddress { get; }
             public System.Collections.Generic.IEnumerable<Akka.IO.Inet.SocketOption> Options { get; }
             public bool PullMode { get; }
+            [System.Runtime.CompilerServices.NullableAttribute(2)]
+            public Akka.IO.TcpSettings TcpSettings { get; set; }
             public override string ToString() { }
         }
         public sealed class Bound : Akka.IO.Tcp.Event
@@ -4033,6 +4035,7 @@ namespace Akka.IO
         public sealed class CompoundWrite : Akka.IO.Tcp.WriteCommand, System.Collections.Generic.IEnumerable<Akka.IO.Tcp.SimpleWriteCommand>, System.Collections.IEnumerable
         {
             public CompoundWrite(Akka.IO.Tcp.SimpleWriteCommand head, Akka.IO.Tcp.WriteCommand tailCommand) { }
+            public override long Bytes { get; }
             public Akka.IO.Tcp.SimpleWriteCommand Head { get; }
             public Akka.IO.Tcp.WriteCommand TailCommand { get; }
             public System.Collections.Generic.IEnumerator<Akka.IO.Tcp.SimpleWriteCommand> GetEnumerator() { }
@@ -4055,6 +4058,8 @@ namespace Akka.IO
             public System.Collections.Generic.IEnumerable<Akka.IO.Inet.SocketOption> Options { get; }
             public bool PullMode { get; }
             public System.Net.EndPoint RemoteAddress { get; }
+            [System.Runtime.CompilerServices.NullableAttribute(2)]
+            public Akka.IO.TcpSettings TcpSettings { get; set; }
             public System.Nullable<System.TimeSpan> Timeout { get; }
             public override string ToString() { }
         }
@@ -4174,6 +4179,7 @@ namespace Akka.IO
         {
             public static readonly Akka.IO.Tcp.Write Empty;
             public override Akka.IO.Tcp.Event Ack { get; }
+            public override long Bytes { get; }
             public Akka.IO.ByteString Data { get; }
             public static Akka.IO.Tcp.Write Create(Akka.IO.ByteString data) { }
             public static Akka.IO.Tcp.Write Create(Akka.IO.ByteString data, Akka.IO.Tcp.Event ack) { }
@@ -4182,6 +4188,7 @@ namespace Akka.IO
         public abstract class WriteCommand : Akka.IO.Tcp.Command
         {
             protected WriteCommand() { }
+            public abstract long Bytes { get; }
             public static Akka.IO.Tcp.WriteCommand Create(System.Collections.Generic.IEnumerable<Akka.IO.Tcp.WriteCommand> writes) { }
             public static Akka.IO.Tcp.WriteCommand Create(params WriteCommand[] writes) { }
             public Akka.IO.Tcp.CompoundWrite Prepend(Akka.IO.Tcp.SimpleWriteCommand other) { }
@@ -4195,7 +4202,6 @@ namespace Akka.IO
     public sealed class TcpExt : Akka.IO.IOExtension
     {
         public TcpExt(Akka.Actor.ExtendedActorSystem system) { }
-        public Akka.IO.Buffers.IBufferPool BufferPool { get; }
         public override Akka.Actor.IActorRef Manager { get; }
         public Akka.IO.TcpSettings Settings { get; }
     }
@@ -4221,21 +4227,31 @@ namespace Akka.IO
         public static Akka.IO.Tcp.Command Unbind() { }
         public static Akka.IO.Tcp.Command Write(Akka.IO.ByteString data, Akka.IO.Tcp.Event ack = null) { }
     }
-    public class TcpSettings
+    public sealed class TcpSettings : System.IEquatable<Akka.IO.TcpSettings>
     {
+        [System.ObsoleteAttribute("Many of these options are no longer used. Use the TcpSettings.Create method inste" +
+            "ad.")]
         public TcpSettings(string bufferPoolConfigPath, int initialSocketAsyncEventArgs, bool traceLogging, int batchAcceptLimit, System.Nullable<System.TimeSpan> registerTimeout, int receivedMessageSizeLimit, string managementDispatcher, string fileIoDispatcher, int transferToLimit, int finishConnectRetries, bool outgoingSocketForceIpv4, int writeCommandsQueueMaxSize) { }
-        public int BatchAcceptLimit { get; }
+        public int BatchAcceptLimit { get; set; }
+        [System.ObsoleteAttribute("This property is unused")]
         public string BufferPoolConfigPath { get; }
+        [System.ObsoleteAttribute("This property is unused")]
         public string FileIODispatcher { get; }
-        public int FinishConnectRetries { get; }
+        public int FinishConnectRetries { get; set; }
+        [System.ObsoleteAttribute("This property is unused")]
         public int InitialSocketAsyncEventArgs { get; }
         public string ManagementDispatcher { get; }
-        public bool OutgoingSocketForceIpv4 { get; }
-        public int ReceivedMessageSizeLimit { get; }
-        public System.Nullable<System.TimeSpan> RegisterTimeout { get; }
-        public bool TraceLogging { get; }
+        public int MaxFrameSizeBytes { get; set; }
+        public bool OutgoingSocketForceIpv4 { get; set; }
+        public int ReceiveBufferSize { get; set; }
+        [System.ObsoleteAttribute("This property is now MaxFrameSizeBytes")]
+        public long ReceivedMessageSizeLimit { get; }
+        public System.Nullable<System.TimeSpan> RegisterTimeout { get; set; }
+        public int SendBufferSize { get; set; }
+        public bool TraceLogging { get; set; }
+        [System.ObsoleteAttribute("This property is unused")]
         public int TransferToLimit { get; set; }
-        public int WriteCommandsQueueMaxSize { get; }
+        public int WriteCommandsQueueMaxSize { get; set; }
         public static Akka.IO.TcpSettings Create(Akka.Actor.ActorSystem system) { }
         public static Akka.IO.TcpSettings Create(Akka.Configuration.Config config) { }
     }

src/core/Akka.Streams.Tests/IO/TcpHelper.cs

@@ -330,9 +330,9 @@ await _connectionProbe.FishForMessageAsync(
 
             public async Task ExpectTerminatedAsync(CancellationToken cancellationToken = default)
             {
-                _connectionProbe.Watch(_connectionActor);
+                await _connectionProbe.WatchAsync(_connectionActor);
                 await _connectionProbe.ExpectTerminatedAsync(_connectionActor, cancellationToken: cancellationToken);
-                _connectionProbe.Unwatch(_connectionActor);
+                await _connectionProbe.UnwatchAsync(_connectionActor);
             }
         }
 

src/core/Akka.Streams.Tests/IO/TcpSpec.cs

@@ -1,4 +1,4 @@
-//-----------------------------------------------------------------------
+//-----------------------------------------------------------------------
 // <copyright file="TcpSpec.cs" company="Akka.NET Project">
 //     Copyright (C) 2009-2022 Lightbend Inc. <http://www.lightbend.com>
 //     Copyright (C) 2013-2025 .NET Foundation <https://github.com/akkadotnet/akka.net>
@@ -29,9 +29,11 @@ namespace Akka.Streams.Tests.IO
 {
     public class TcpSpec : TcpHelper
     {
-        public TcpSpec(ITestOutputHelper helper) : base(@"
-akka.loglevel = DEBUG
-akka.stream.materializer.subscription-timeout.timeout = 2s", helper)
+        public TcpSpec(ITestOutputHelper helper) : base("""
+                                                        akka.io.tcp.trace-logging = on
+                                                        akka.loglevel = DEBUG
+                                                        akka.stream.materializer.subscription-timeout.timeout = 2s
+                                                        """, helper)
         {
         }
 
@@ -40,7 +42,7 @@ public async Task Outgoing_TCP_stream_must_work_in_the_happy_case()
         {
             await this.AssertAllStagesStoppedAsync(async () =>
             {
-                var testData = ByteString.FromBytes(new byte[] {1, 2, 3, 4, 5});
+                var testData = ByteString.FromBytes([1, 2, 3, 4, 5]);
 
                 var server = await new Server(this).InitializeAsync();
                 var tcpReadProbe = new TcpReadProbe(this);
@@ -65,7 +67,7 @@ await ValidateServerClientCommunicationAsync(testData, serverConnection, tcpRead
         public async Task Outgoing_TCP_stream_must_be_able_to_write_a_sequence_of_ByteStrings()
         {
             var server = await new Server(this).InitializeAsync();
-            var testInput = Enumerable.Range(0, 256).Select(i => ByteString.FromBytes(new byte[] {Convert.ToByte(i)}));
+            var testInput = Enumerable.Range(0, 256).Select(i => ByteString.FromBytes([Convert.ToByte(i)]));
             var expectedOutput = ByteString.FromBytes(Enumerable.Range(0, 256).Select(Convert.ToByte).ToArray());
 
             Source.From(testInput)
@@ -86,7 +88,7 @@ public async Task Outgoing_TCP_stream_must_be_able_to_read_a_sequence_of_ByteStr
             var testOutput = new byte[255];
             for (byte i = 0; i < 255; i++)
             {
-                testInput[i] = ByteString.FromBytes(new [] {i});
+                testInput[i] = ByteString.FromBytes([i]);
                 testOutput[i] = i;
             }
 
@@ -103,6 +105,7 @@ public async Task Outgoing_TCP_stream_must_be_able_to_read_a_sequence_of_ByteStr
                 serverConnection.Write(input);
 
             serverConnection.ConfirmedClose();
+            // Reduced timeout - otherwise we're just waiting longer for the failure
             var result = await resultFuture.ShouldCompleteWithin(3.Seconds());
             result.ShouldBe(expectedOutput);
         }
@@ -132,7 +135,7 @@ public async Task Outgoing_TCP_stream_must_work_when_client_closes_write_then_re
         {
             await this.AssertAllStagesStoppedAsync(async () =>
             {
-                var testData = ByteString.FromBytes(new byte[] { 1, 2, 3, 4, 5 });
+                var testData = ByteString.FromBytes([1, 2, 3, 4, 5]);
                 var server = await new Server(this).InitializeAsync();
 
                 var tcpWriteProbe = new TcpWriteProbe(this);
@@ -466,7 +469,7 @@ public async Task Outgoing_TCP_stream_must_Echo_should_work_even_if_server_is_in
                     .Run(Materializer);
 
             var result = await Source.From(Enumerable.Repeat(0, 1000)
-                .Select(i => ByteString.FromBytes(new byte[] { Convert.ToByte(i) })))
+                .Select(i => ByteString.FromBytes([Convert.ToByte(i)])))
                 .Via(Sys.TcpStream().OutgoingConnection(serverAddress, halfClose: true))
                 .RunAggregate(0, (i, s) => i + s.Count, Materializer).ShouldCompleteWithin(10.Seconds());
 
@@ -503,7 +506,7 @@ await WithinAsync(TimeSpan.FromSeconds(15), async () =>
                             await AwaitAssertAsync(async () =>
                             {
                                 // Getting rid of existing connection actors by using a blunt instrument
-                                system2.ActorSelection(system2.Tcp().Path / "$a" / "*").Tell(Kill.Instance);
+                                system2.ActorSelection(system2.Tcp().Path / "tcp-client-connection-*").Tell(Kill.Instance);
 
                                 await result.ShouldCompleteWithin(3.Seconds());
                             }, interval:TimeSpan.FromSeconds(4));
@@ -525,6 +528,7 @@ await AwaitAssertAsync(async () =>
         public async Task Outgoing_TCP_stream_must_not_thrown_on_unbind_after_system_has_been_shut_down()
         {
             var sys2 = ActorSystem.Create("shutdown-test-system", Sys.Settings.Config);
+            InitializeLogger(sys2);
 
             try
             {
@@ -573,7 +577,7 @@ public async Task Tcp_listen_stream_must_be_able_to_implement_echo()
             var binding = await bindTask.ShouldCompleteWithin(3.Seconds());
 
             var testInput = Enumerable.Range(0, 255)
-                .Select(i => ByteString.FromBytes(new byte[] { Convert.ToByte(i) }))
+                .Select(i => ByteString.FromBytes([Convert.ToByte(i)]))
                 .ToList();
 
             var expectedOutput = testInput.Aggregate(ByteString.Empty, (agg, b) => agg.Concat(b));
@@ -603,7 +607,7 @@ public async Task Tcp_listen_stream_must_work_with_a_chain_of_echoes()
             var echoConnection = Sys.TcpStream().OutgoingConnection(serverAddress);
 
             var testInput = Enumerable.Range(0, 255)
-                .Select(i => ByteString.FromBytes(new byte[] { Convert.ToByte(i) }))
+                .Select(i => ByteString.FromBytes([Convert.ToByte(i)]))
                 .ToList();
 
             var expectedOutput = testInput.Aggregate(ByteString.Empty, (agg, b) => agg.Concat(b));

src/core/Akka.Streams/Dsl/Tcp.cs

@@ -180,13 +180,14 @@ public TcpExt(ExtendedActorSystem system)
         /// <param name="idleTimeout">TBD</param>
         /// <exception cref="ArgumentException">TBD</exception>
         /// <returns>TBD</returns>
+        // TODO: this really needs to be an async method
         public Source<Tcp.IncomingConnection, Task<Tcp.ServerBinding>> Bind(string host, int port, int backlog = 100,
             IImmutableList<Inet.SocketOption> options = null, bool halfClose = false, TimeSpan? idleTimeout = null)
         {
             // DnsEndpoint isn't allowed
             var ipAddresses = System.Net.Dns.GetHostAddressesAsync(host).Result;
             if (ipAddresses.Length == 0)
-                throw new ArgumentException($"Couldn't resolve IpAdress for host {host}", nameof(host));
+                throw new ArgumentException($"Couldn't resolve IpAddress for host {host}", nameof(host));
 
             return Source.FromGraph(new ConnectionSourceStage(_system.Tcp(), new IPEndPoint(ipAddresses[0], port), backlog,
                 options, halfClose, idleTimeout, BindShutdownTimeout));