commit 8c0479a3b8fa784ba1988e8948a359b927664bf3
parent 8f7f63eb48376ec161dec362707e04dcaed88afc
Author: Mikolaj Lenczewski <33129490+EnderRifter@users.noreply.github.com>
Date: Fri, 28 Feb 2020 14:30:59 +0000
Added instance builders and basic implementation of a packet pipeline.
Diffstat:
13 files changed, 830 insertions(+), 148 deletions(-)
diff --git a/NetSharp/NetSharp/Connection.cs b/NetSharp/NetSharp/Connection.cs
@@ -7,6 +7,7 @@ using System.Threading;
using System.Threading.Tasks;
using NetSharp.Logging;
using NetSharp.Packets;
+using NetSharp.Pipelines;
using NetSharp.Sockets;
using NetSharp.Utils;
@@ -17,7 +18,9 @@ namespace NetSharp
private readonly SocketAcceptor acceptor;
private readonly ConcurrentDictionary<EndPoint, Connection> connections;
private readonly CancellationTokenSource connectionShutdownTokenSource;
+ private readonly PacketPipeline<Memory<byte>, Memory<byte>, NetworkPacket> incomingPacketPipeline;
private readonly SocketReader listener;
+ private readonly PacketPipeline<NetworkPacket, Memory<byte>, Memory<byte>> outgoingPacketPipeline;
private readonly Socket socket;
private readonly SocketWriter transmitter;
@@ -29,6 +32,41 @@ namespace NetSharp
Dispose(false);
}
+ private void AcceptorWork(object shutdownToken)
+ {
+ CancellationToken cancellationToken = (CancellationToken)shutdownToken;
+
+ while (!cancellationToken.IsCancellationRequested)
+ {
+ }
+ }
+
+ private void ListenerWork(object shutdownToken)
+ {
+ CancellationToken cancellationToken = (CancellationToken)shutdownToken;
+
+ while (!cancellationToken.IsCancellationRequested)
+ {
+ }
+ }
+
+ private void PacketPipelineWork(object shutdownToken)
+ {
+ CancellationToken cancellationToken = (CancellationToken)shutdownToken;
+
+ while (!cancellationToken.IsCancellationRequested)
+ {
+ }
+ }
+
+ /// <summary>
+ /// Lock synchronisation object for the <see cref="logger"/> variable.
+ /// </summary>
+ protected readonly object loggerLockObject = new object();
+
+ /// <summary>
+ /// Cancellation token which allows observing the shutdown of the server. It is set when <see cref="Shutdown()"/> is called.
+ /// </summary>
protected readonly CancellationToken ShutdownToken;
/// <summary>
@@ -49,8 +87,10 @@ namespace NetSharp
}
internal Connection(AddressFamily addressFamily, SocketType socketType, ProtocolType protocolType,
- int objectPoolSize = 10, bool preallocateBuffers = false, Stream? loggingStream = default,
- LogLevel minimumLoggedSeverity = LogLevel.Info)
+ PacketPipeline<Memory<byte>, Memory<byte>, NetworkPacket> incomingPacketPipeline,
+ PacketPipeline<NetworkPacket, Memory<byte>, Memory<byte>> outgoingPacketPipeline,
+ int objectPoolSize = 10, bool preallocateBuffers = false, Stream? loggingStream = default,
+ LogLevel minimumLoggedSeverity = LogLevel.Info)
{
connectionShutdownTokenSource = new CancellationTokenSource();
ShutdownToken = connectionShutdownTokenSource.Token;
@@ -63,6 +103,9 @@ namespace NetSharp
connections = new ConcurrentDictionary<EndPoint, Connection>();
+ this.incomingPacketPipeline = incomingPacketPipeline;
+ this.outgoingPacketPipeline = outgoingPacketPipeline;
+
logger = new Logger(loggingStream ?? Stream.Null, minimumLoggedSeverity);
}
@@ -73,9 +116,6 @@ namespace NetSharp
GC.SuppressFinalize(this);
}
- public Task<TransmissionResult> ReceiveAsync(Memory<byte> inputBuffer, SocketFlags flags)
- => ReceiveAsync(inputBuffer, flags, Timeout.InfiniteTimeSpan);
-
public Task<TransmissionResult> ReceiveAsync(Memory<byte> inputBuffer, SocketFlags flags, TimeSpan timeout)
{
using CancellationTokenSource timeoutCancellationTokenSource = new CancellationTokenSource(timeout);
@@ -85,9 +125,6 @@ namespace NetSharp
return listener.ReceiveAsync(socket, flags, inputBuffer, cts.Token);
}
- public Task<TransmissionResult> ReceiveFromAsync(EndPoint remoteEndPoint, Memory<byte> inputBuffer, SocketFlags flags)
- => ReceiveFromAsync(remoteEndPoint, inputBuffer, flags, Timeout.InfiniteTimeSpan);
-
public Task<TransmissionResult> ReceiveFromAsync(EndPoint remoteEndPoint, Memory<byte> inputBuffer, SocketFlags flags, TimeSpan timeout)
{
using CancellationTokenSource timeoutCancellationTokenSource = new CancellationTokenSource(timeout);
@@ -97,8 +134,27 @@ namespace NetSharp
return listener.ReceiveFromAsync(socket, remoteEndPoint, flags, inputBuffer, cts.Token);
}
- public Task<int> SendAsync(Memory<byte> outputBuffer, SocketFlags flags)
- => SendAsync(outputBuffer, flags, Timeout.InfiniteTimeSpan);
+ /// <summary>
+ /// Makes the connection listen for incoming client request packets, and handle them according to registered packet handler delegates.
+ /// This work can be cancelled by calling <see cref="Shutdown()"/>.
+ /// </summary>
+ /// <returns>The task representing the connection work.</returns>
+ public Task RunAsync()
+ {
+ Task acceptorThread =
+ Task.Factory.StartNew(AcceptorWork, ShutdownToken, ShutdownToken, TaskCreationOptions.LongRunning,
+ TaskScheduler.Default);
+
+ Task listenerThread =
+ Task.Factory.StartNew(ListenerWork, ShutdownToken, ShutdownToken, TaskCreationOptions.LongRunning,
+ TaskScheduler.Default);
+
+ Task packetPipelineThread =
+ Task.Factory.StartNew(PacketPipelineWork, ShutdownToken, ShutdownToken, TaskCreationOptions.LongRunning,
+ TaskScheduler.Default);
+
+ return Task.WhenAll(acceptorThread, listenerThread, packetPipelineThread);
+ }
public Task<int> SendAsync(Memory<byte> outputBuffer, SocketFlags flags, TimeSpan timeout)
{
@@ -109,9 +165,6 @@ namespace NetSharp
return transmitter.SendAsync(socket, flags, outputBuffer, cts.Token);
}
- public Task<int> SendToAsync(EndPoint remoteEndPoint, Memory<byte> outputBuffer, SocketFlags flags)
- => SendToAsync(remoteEndPoint, outputBuffer, flags, Timeout.InfiniteTimeSpan);
-
public Task<int> SendToAsync(EndPoint remoteEndPoint, Memory<byte> outputBuffer, SocketFlags flags, TimeSpan timeout)
{
using CancellationTokenSource timeoutCancellationTokenSource = new CancellationTokenSource(timeout);
@@ -121,9 +174,26 @@ namespace NetSharp
return transmitter.SendToAsync(socket, remoteEndPoint, flags, outputBuffer, cts.Token);
}
+ /// <summary>
+ /// Configures the logger to log messages to the given stream (or to <see cref="Stream.Null"/> if <c>null</c>) and
+ /// to only log messages that are of severity <paramref name="minimumLoggedSeverity"/> or higher.
+ /// </summary>
+ /// <param name="loggingStream">The stream to which messages will be logged.</param>
+ /// <param name="minimumLoggedSeverity">The minimum severity a message must be to be logged.</param>
public void SetLoggingStream(Stream? loggingStream, LogLevel minimumLoggedSeverity = LogLevel.Info)
{
- logger = new Logger(loggingStream ?? Stream.Null, minimumLoggedSeverity);
+ lock (loggerLockObject)
+ {
+ logger = new Logger(loggingStream ?? Stream.Null, minimumLoggedSeverity);
+ }
+ }
+
+ /// <summary>
+ /// Shuts down the connection, and releases managed and unmanaged resources.
+ /// </summary>
+ public void Shutdown()
+ {
+ connectionShutdownTokenSource.Cancel();
}
/// <summary>
diff --git a/NetSharp/NetSharp/ConnectionBuilder.cs b/NetSharp/NetSharp/ConnectionBuilder.cs
@@ -0,0 +1,244 @@
+using System;
+using System.Collections.Generic;
+using System.IO;
+using System.Net.Sockets;
+using NetSharp.Logging;
+using NetSharp.Packets;
+using NetSharp.Pipelines;
+
+namespace NetSharp
+{
+ /// <summary>
+ /// Allows for configuring and subsequently building a <see cref="Connection"/> instance.
+ /// </summary>
+ public sealed class ConnectionBuilder
+ {
+ private static readonly LoggingSettings DefaultLoggingSettings = new LoggingSettings(Stream.Null, LogLevel.Warn);
+ private static readonly PoolingSettings DefaultPoolingSettings = new PoolingSettings(10, false);
+
+ private readonly List<Func<Memory<byte>, Memory<byte>>> incomingPipelineStages =
+ new List<Func<Memory<byte>, Memory<byte>>>();
+
+ private readonly List<Func<Memory<byte>, Memory<byte>>> outgoingPipelineStages =
+ new List<Func<Memory<byte>, Memory<byte>>>();
+
+ private LoggingSettings? loggingSettings;
+ private PoolingSettings? poolingSettings;
+ private SocketSettings? socketSettings;
+
+ /// <summary>
+ /// The number of stages in the currently configured incoming packet pipeline.
+ /// </summary>
+ public int IncomingPacketPipelineStageCount
+ {
+ get { return incomingPipelineStages.Count; }
+ }
+
+ /// <summary>
+ /// The number of stages in the currently configured outgoing packet pipeline.
+ /// </summary>
+ public int OutgoingPacketPipelineStageCount
+ {
+ get { return outgoingPipelineStages.Count; }
+ }
+
+ /// <summary>
+ /// Returns a new <see cref="Connection"/> instance with the current configuration.
+ /// </summary>
+ /// <returns>The configured <see cref="Connection"/> instance.</returns>
+ /// <exception cref="ArgumentNullException">
+ /// Thrown when <see cref="WithSocket"/> has not been called.
+ /// </exception>
+ public Connection Build()
+ {
+ if (socketSettings == null)
+ {
+ throw new ArgumentNullException(nameof(socketSettings), $"{nameof(WithSocket)} has not been called.");
+ }
+
+ PacketPipelineBuilder<Memory<byte>, Memory<byte>, NetworkPacket> incomingPipelineBuilder =
+ new PacketPipelineBuilder<Memory<byte>, Memory<byte>, NetworkPacket>();
+
+ incomingPipelineBuilder.WithInputStage(memory => memory);
+ foreach (Func<Memory<byte>, Memory<byte>> stage in incomingPipelineStages)
+ {
+ incomingPipelineBuilder = incomingPipelineBuilder.WithIntermediateStage(stage);
+ }
+
+ incomingPipelineBuilder.WithOutputStage(NetworkPacket.Deserialise);
+
+ PacketPipelineBuilder<NetworkPacket, Memory<byte>, Memory<byte>> outgoingPipelineBuilder =
+ new PacketPipelineBuilder<NetworkPacket, Memory<byte>, Memory<byte>>();
+
+ outgoingPipelineBuilder.WithInputStage(NetworkPacket.Serialise);
+ foreach (Func<Memory<byte>, Memory<byte>> stage in outgoingPipelineStages)
+ {
+ outgoingPipelineBuilder = outgoingPipelineBuilder.WithIntermediateStage(stage);
+ }
+
+ outgoingPipelineBuilder.WithOutputStage(memory => memory);
+
+ Connection connection = new Connection(
+ socketSettings.Value.AddressFamily,
+ socketSettings.Value.SocketType,
+ socketSettings.Value.ProtocolType,
+ incomingPipelineBuilder.Build(),
+ outgoingPipelineBuilder.Build(),
+ poolingSettings?.ObjectPoolSize ?? DefaultPoolingSettings.ObjectPoolSize,
+ poolingSettings?.PreallocateBuffers ?? DefaultPoolingSettings.PreallocateBuffers,
+ loggingSettings?.LoggingStream ?? DefaultLoggingSettings.LoggingStream,
+ loggingSettings?.MinimumLevel ?? DefaultLoggingSettings.MinimumLevel);
+
+ return connection;
+ }
+
+ /// <summary>
+ /// Adds an extra pipeline stage to the currently configured incoming packet pipeline, at the given index.
+ /// </summary>
+ /// <param name="transform">
+ /// The transformation that should be applied when a packet passes through the pipeline.
+ /// </param>
+ /// <param name="index">The position in the pipeline at which to place the transform.</param>
+ /// <returns>The builder instance for further configuration.</returns>
+ public ConnectionBuilder WithIncomingPipelineStage(in Func<Memory<byte>, Memory<byte>> transform, int index)
+ {
+ incomingPipelineStages.Insert(index, transform);
+ return this;
+ }
+
+ /// <summary>
+ /// Sets the logging settings for the currently configured connection.
+ /// </summary>
+ /// <param name="settings">The logging settings to use.</param>
+ /// <returns>The builder instance for further configuration.</returns>
+ public ConnectionBuilder WithLogging(LoggingSettings settings)
+ {
+ loggingSettings = settings;
+ return this;
+ }
+
+ /// <summary>
+ /// Adds an extra pipeline stage to the currently configured outgoing packet pipeline, at the given index.
+ /// </summary>
+ /// <param name="transform">
+ /// The transformation that should be applied when a packet passes through the pipeline.
+ /// </param>
+ /// <param name="index">The position in the pipeline at which to place the transform.</param>
+ /// <returns>The builder instance for further configuration.</returns>
+ public ConnectionBuilder WithOutgoingPipelineStage(in Func<Memory<byte>, Memory<byte>> transform, int index)
+ {
+ outgoingPipelineStages.Insert(index, transform);
+ return this;
+ }
+
+ /// <summary>
+ /// Sets the pooling settings for the currently configured connection.
+ /// </summary>
+ /// <param name="settings">The pooling settings to use.</param>
+ /// <returns>The builder instance for further configuration.</returns>
+ public ConnectionBuilder WithPooling(PoolingSettings settings)
+ {
+ poolingSettings = settings;
+ return this;
+ }
+
+ /// <summary>
+ /// Sets the socket settings for the currently configured connection.
+ /// </summary>
+ /// <param name="settings">The socket settings to use.</param>
+ /// <returns>The builder instance for further configuration.</returns>
+ public ConnectionBuilder WithSocket(SocketSettings settings)
+ {
+ socketSettings = settings;
+ return this;
+ }
+
+ /// <summary>
+ /// Holds settings for configuring a connection's logging.
+ /// </summary>
+ public readonly struct LoggingSettings
+ {
+ /// <summary>
+ /// The stream to which messages will be logged.
+ /// </summary>
+ public readonly Stream LoggingStream;
+
+ /// <summary>
+ /// The minimum severity that a log message must have to be recorded.
+ /// </summary>
+ public readonly LogLevel MinimumLevel;
+
+ /// <summary>
+ /// Initialises a new instance of the <see cref="LoggingStream"/> struct.
+ /// </summary>
+ /// <param name="stream">The stream to which messages will be logged..</param>
+ /// <param name="minimumLevel">The minimum severity that a log message must have to be recorded.</param>
+ public LoggingSettings(Stream stream, LogLevel minimumLevel)
+ {
+ LoggingStream = stream;
+ MinimumLevel = minimumLevel;
+ }
+ }
+
+ /// <summary>
+ /// Holds settings for configuring a connection's buffer pooling.
+ /// </summary>
+ public readonly struct PoolingSettings
+ {
+ /// <summary>
+ /// The number of objects that will be held in the object pools.
+ /// </summary>
+ public readonly int ObjectPoolSize;
+
+ /// <summary>
+ /// Whether the buffers for receiving messages should be preallocated.
+ /// </summary>
+ public readonly bool PreallocateBuffers;
+
+ /// <summary>
+ /// Initialises a new instance of the <see cref="PoolingSettings"/> struct.
+ /// </summary>
+ /// <param name="poolSize">The number of objects that will be held in the object pools.</param>
+ /// <param name="preallocateBuffers">Whether the buffers for receiving messages should be preallocated.</param>
+ public PoolingSettings(int poolSize, bool preallocateBuffers)
+ {
+ ObjectPoolSize = poolSize;
+ PreallocateBuffers = preallocateBuffers;
+ }
+ }
+
+ /// <summary>
+ /// Holds settings fo configuring a connection's underlying socket.
+ /// </summary>
+ public readonly struct SocketSettings
+ {
+ /// <summary>
+ /// The address family for the socket underlying the connection.
+ /// </summary>
+ public readonly AddressFamily AddressFamily;
+
+ /// <summary>
+ /// The socket type for the socket underlying the connection.
+ /// </summary>
+ public readonly ProtocolType ProtocolType;
+
+ /// <summary>
+ /// The protocol type for the socket underlying the connection.
+ /// </summary>
+ public readonly SocketType SocketType;
+
+ /// <summary>
+ /// Initialises a new instance of the <see cref="SocketSettings"/> struct.
+ /// </summary>
+ /// <param name="addressFamily">The address family for the underlying socket.</param>
+ /// <param name="socketType">The socket type for the underlying socket.</param>
+ /// <param name="protocolType">The protocol type for the underlying socket.</param>
+ public SocketSettings(AddressFamily addressFamily, SocketType socketType, ProtocolType protocolType)
+ {
+ AddressFamily = addressFamily;
+ SocketType = socketType;
+ ProtocolType = protocolType;
+ }
+ }
+ }
+}
+\ No newline at end of file
diff --git a/NetSharp/NetSharp/ConnectionFactory.cs b/NetSharp/NetSharp/ConnectionFactory.cs
@@ -1,36 +0,0 @@
-using System;
-using System.IO;
-using System.Net;
-using System.Net.Sockets;
-using NetSharp.Logging;
-
-namespace NetSharp
-{
- /// <summary>
- /// Provides methods to construct and configure <see cref="Connection"/> instances.
- /// </summary>
- public static class ConnectionFactory
- {
- public static Connection Init(EndPoint localEndPoint, AddressFamily connectionAddressFamily,
- SocketType connectionSocketType, ProtocolType connectionProtocolType, TimeSpan timeout, int objectPoolSize = 10,
- bool preallocateBuffers = false, Stream? loggingStream = default, LogLevel minimumLoggedSeverity = LogLevel.Info)
- {
- Connection connection = new Connection(connectionAddressFamily, connectionSocketType,
- connectionProtocolType, objectPoolSize, preallocateBuffers, loggingStream, minimumLoggedSeverity);
-
- connection.TryBind(localEndPoint, timeout);
-
- return connection;
- }
-
- public static Connection InitTcp(EndPoint remoteEndPoint, int objectPoolSize = 10, bool preallocateBuffers = false,
- Stream? loggingStream = default, LogLevel minimumLoggedSeverity = LogLevel.Info)
- => Init(remoteEndPoint, AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp, TimeSpan.FromSeconds(1),
- objectPoolSize, preallocateBuffers, loggingStream, minimumLoggedSeverity);
-
- public static Connection InitUdp(EndPoint remoteEndPoint, int objectPoolSize = 10, bool preallocateBuffers = false,
- Stream? loggingStream = default, LogLevel minimumLoggedSeverity = LogLevel.Info)
- => Init(remoteEndPoint, AddressFamily.InterNetwork, SocketType.Dgram, ProtocolType.Udp, TimeSpan.FromSeconds(1),
- objectPoolSize, preallocateBuffers, loggingStream, minimumLoggedSeverity);
- }
-}
-\ No newline at end of file
diff --git a/NetSharp/NetSharp/Extensions/ConnectionBuilderExtensions.cs b/NetSharp/NetSharp/Extensions/ConnectionBuilderExtensions.cs
@@ -0,0 +1,39 @@
+using System;
+using System.IO;
+using System.Net.Sockets;
+using NetSharp.Logging;
+
+namespace NetSharp.Extensions
+{
+ /// <summary>
+ /// Provides additional methods and functionality to the <see cref="ConnectionBuilder"/> class.
+ /// </summary>
+ public static class ConnectionBuilderExtensions
+ {
+ public static ConnectionBuilder AppendIncomingPipelineStage(this ConnectionBuilder instance,
+ in Func<Memory<byte>, Memory<byte>> transform)
+ => instance.WithIncomingPipelineStage(transform, instance.IncomingPacketPipelineStageCount);
+
+ public static ConnectionBuilder AppendOutgoingPipelineStage(this ConnectionBuilder instance,
+ in Func<Memory<byte>, Memory<byte>> transform)
+ => instance.WithOutgoingPipelineStage(transform, instance.OutgoingPacketPipelineStageCount);
+
+ public static ConnectionBuilder WithLogging(this ConnectionBuilder instance,
+ Stream loggingStream, LogLevel minimumLogLevel)
+ => instance.WithLogging(new ConnectionBuilder.LoggingSettings(loggingStream, minimumLogLevel));
+
+ public static ConnectionBuilder WithPooling(this ConnectionBuilder instance,
+ int poolSize, bool preallocateBuffers)
+ => instance.WithPooling(new ConnectionBuilder.PoolingSettings(poolSize, preallocateBuffers));
+
+ public static ConnectionBuilder WithSocket(this ConnectionBuilder instance,
+ AddressFamily addressFamily, SocketType socketType, ProtocolType protocolType)
+ => instance.WithSocket(new ConnectionBuilder.SocketSettings(addressFamily, socketType, protocolType));
+
+ public static ConnectionBuilder WithTcp(this ConnectionBuilder instance)
+ => instance.WithSocket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
+
+ public static ConnectionBuilder WithUdp(this ConnectionBuilder instance)
+ => instance.WithSocket(AddressFamily.InterNetwork, SocketType.Dgram, ProtocolType.Udp);
+ }
+}
+\ No newline at end of file
diff --git a/NetSharp/NetSharp/Extensions/ConnectionExtensions.cs b/NetSharp/NetSharp/Extensions/ConnectionExtensions.cs
@@ -12,5 +12,28 @@ namespace NetSharp.Extensions
/// </summary>
public static class ConnectionExtensions
{
+ public static Task<TransmissionResult> ReceiveAsync(this Connection instance,
+ Memory<byte> inputBuffer, SocketFlags flags)
+ => instance.ReceiveAsync(inputBuffer, flags, Timeout.InfiniteTimeSpan);
+
+ public static Task<TransmissionResult> ReceiveFromAsync(this Connection instance,
+ EndPoint remoteEndPoint, Memory<byte> inputBuffer, SocketFlags flags)
+ => instance.ReceiveFromAsync(remoteEndPoint, inputBuffer, flags, Timeout.InfiniteTimeSpan);
+
+ public static Task<int> SendAsync(this Connection instance,
+ Memory<byte> outputBuffer, SocketFlags flags)
+ => instance.SendAsync(outputBuffer, flags, Timeout.InfiniteTimeSpan);
+
+ public static Task<int> SendToAsync(this Connection instance,
+ EndPoint remoteEndPoint, Memory<byte> outputBuffer, SocketFlags flags)
+ => instance.SendToAsync(remoteEndPoint, outputBuffer, flags, Timeout.InfiniteTimeSpan);
+
+ public static bool TryBind(this Connection instance,
+ EndPoint localEndPoint)
+ => instance.TryBind(localEndPoint, Timeout.InfiniteTimeSpan);
+
+ public static Task<bool> TryBindAsync(this Connection instance,
+ EndPoint localEndPoint)
+ => instance.TryBindAsync(localEndPoint, Timeout.InfiniteTimeSpan);
}
}
\ No newline at end of file
diff --git a/NetSharp/NetSharp/NetSharp.xml b/NetSharp/NetSharp/NetSharp.xml
@@ -9,6 +9,16 @@
Destroys a <see cref="T:NetSharp.Connection"/> class instance, freeing all managed resources.
</summary>
</member>
+ <member name="F:NetSharp.Connection.loggerLockObject">
+ <summary>
+ Lock synchronisation object for the <see cref="F:NetSharp.Connection.logger"/> variable.
+ </summary>
+ </member>
+ <member name="F:NetSharp.Connection.ShutdownToken">
+ <summary>
+ Cancellation token which allows observing the shutdown of the server. It is set when <see cref="M:NetSharp.Connection.Shutdown"/> is called.
+ </summary>
+ </member>
<member name="F:NetSharp.Connection.logger">
<summary>
A logger object allowing for writing debug messages to an output stream.
@@ -23,6 +33,26 @@
<member name="M:NetSharp.Connection.Dispose">
<inheritdoc />
</member>
+ <member name="M:NetSharp.Connection.RunAsync">
+ <summary>
+ Makes the connection listen for incoming client request packets, and handle them according to registered packet handler delegates.
+ This work can be cancelled by calling <see cref="M:NetSharp.Connection.Shutdown"/>.
+ </summary>
+ <returns>The task representing the connection work.</returns>
+ </member>
+ <member name="M:NetSharp.Connection.SetLoggingStream(System.IO.Stream,NetSharp.Logging.LogLevel)">
+ <summary>
+ Configures the logger to log messages to the given stream (or to <see cref="F:System.IO.Stream.Null"/> if <c>null</c>) and
+ to only log messages that are of severity <paramref name="minimumLoggedSeverity"/> or higher.
+ </summary>
+ <param name="loggingStream">The stream to which messages will be logged.</param>
+ <param name="minimumLoggedSeverity">The minimum severity a message must be to be logged.</param>
+ </member>
+ <member name="M:NetSharp.Connection.Shutdown">
+ <summary>
+ Shuts down the connection, and releases managed and unmanaged resources.
+ </summary>
+ </member>
<member name="M:NetSharp.Connection.TryBind(System.Net.EndPoint,System.TimeSpan)">
<summary>
Attempts to synchronously bind the underlying socket to the given local endpoint. Blocks.
@@ -41,11 +71,143 @@
<param name="timeout">The timeout within which to attempt the binding.</param>
<returns>Whether the binding was successful or not.</returns>
</member>
- <member name="T:NetSharp.ConnectionFactory">
+ <member name="T:NetSharp.ConnectionBuilder">
+ <summary>
+ Allows for configuring and subsequently building a <see cref="T:NetSharp.Connection"/> instance.
+ </summary>
+ </member>
+ <member name="P:NetSharp.ConnectionBuilder.IncomingPacketPipelineStageCount">
+ <summary>
+ The number of stages in the currently configured incoming packet pipeline.
+ </summary>
+ </member>
+ <member name="P:NetSharp.ConnectionBuilder.OutgoingPacketPipelineStageCount">
+ <summary>
+ The number of stages in the currently configured outgoing packet pipeline.
+ </summary>
+ </member>
+ <member name="M:NetSharp.ConnectionBuilder.Build">
+ <summary>
+ Returns a new <see cref="T:NetSharp.Connection"/> instance with the current configuration.
+ </summary>
+ <returns>The configured <see cref="T:NetSharp.Connection"/> instance.</returns>
+ <exception cref="T:System.ArgumentNullException">
+ Thrown when <see cref="M:NetSharp.ConnectionBuilder.WithSocket(NetSharp.ConnectionBuilder.SocketSettings)"/> has not been called.
+ </exception>
+ </member>
+ <member name="M:NetSharp.ConnectionBuilder.WithIncomingPipelineStage(System.Func{System.Memory{System.Byte},System.Memory{System.Byte}}@,System.Int32)">
+ <summary>
+ Adds an extra pipeline stage to the currently configured incoming packet pipeline, at the given index.
+ </summary>
+ <param name="transform">
+ The transformation that should be applied when a packet passes through the pipeline.
+ </param>
+ <param name="index">The position in the pipeline at which to place the transform.</param>
+ <returns>The builder instance for further configuration.</returns>
+ </member>
+ <member name="M:NetSharp.ConnectionBuilder.WithLogging(NetSharp.ConnectionBuilder.LoggingSettings)">
+ <summary>
+ Sets the logging settings for the currently configured connection.
+ </summary>
+ <param name="settings">The logging settings to use.</param>
+ <returns>The builder instance for further configuration.</returns>
+ </member>
+ <member name="M:NetSharp.ConnectionBuilder.WithOutgoingPipelineStage(System.Func{System.Memory{System.Byte},System.Memory{System.Byte}}@,System.Int32)">
+ <summary>
+ Adds an extra pipeline stage to the currently configured outgoing packet pipeline, at the given index.
+ </summary>
+ <param name="transform">
+ The transformation that should be applied when a packet passes through the pipeline.
+ </param>
+ <param name="index">The position in the pipeline at which to place the transform.</param>
+ <returns>The builder instance for further configuration.</returns>
+ </member>
+ <member name="M:NetSharp.ConnectionBuilder.WithPooling(NetSharp.ConnectionBuilder.PoolingSettings)">
+ <summary>
+ Sets the pooling settings for the currently configured connection.
+ </summary>
+ <param name="settings">The pooling settings to use.</param>
+ <returns>The builder instance for further configuration.</returns>
+ </member>
+ <member name="M:NetSharp.ConnectionBuilder.WithSocket(NetSharp.ConnectionBuilder.SocketSettings)">
+ <summary>
+ Sets the socket settings for the currently configured connection.
+ </summary>
+ <param name="settings">The socket settings to use.</param>
+ <returns>The builder instance for further configuration.</returns>
+ </member>
+ <member name="T:NetSharp.ConnectionBuilder.LoggingSettings">
+ <summary>
+ Holds settings for configuring a connection's logging.
+ </summary>
+ </member>
+ <member name="F:NetSharp.ConnectionBuilder.LoggingSettings.LoggingStream">
+ <summary>
+ The stream to which messages will be logged.
+ </summary>
+ </member>
+ <member name="F:NetSharp.ConnectionBuilder.LoggingSettings.MinimumLevel">
+ <summary>
+ The minimum severity that a log message must have to be recorded.
+ </summary>
+ </member>
+ <member name="M:NetSharp.ConnectionBuilder.LoggingSettings.#ctor(System.IO.Stream,NetSharp.Logging.LogLevel)">
+ <summary>
+ Initialises a new instance of the <see cref="F:NetSharp.ConnectionBuilder.LoggingSettings.LoggingStream"/> struct.
+ </summary>
+ <param name="stream">The stream to which messages will be logged..</param>
+ <param name="minimumLevel">The minimum severity that a log message must have to be recorded.</param>
+ </member>
+ <member name="T:NetSharp.ConnectionBuilder.PoolingSettings">
<summary>
- Provides methods to construct and configure <see cref="T:NetSharp.Connection"/> instances.
+ Holds settings for configuring a connection's buffer pooling.
</summary>
</member>
+ <member name="F:NetSharp.ConnectionBuilder.PoolingSettings.ObjectPoolSize">
+ <summary>
+ The number of objects that will be held in the object pools.
+ </summary>
+ </member>
+ <member name="F:NetSharp.ConnectionBuilder.PoolingSettings.PreallocateBuffers">
+ <summary>
+ Whether the buffers for receiving messages should be preallocated.
+ </summary>
+ </member>
+ <member name="M:NetSharp.ConnectionBuilder.PoolingSettings.#ctor(System.Int32,System.Boolean)">
+ <summary>
+ Initialises a new instance of the <see cref="T:NetSharp.ConnectionBuilder.PoolingSettings"/> struct.
+ </summary>
+ <param name="poolSize">The number of objects that will be held in the object pools.</param>
+ <param name="preallocateBuffers">Whether the buffers for receiving messages should be preallocated.</param>
+ </member>
+ <member name="T:NetSharp.ConnectionBuilder.SocketSettings">
+ <summary>
+ Holds settings fo configuring a connection's underlying socket.
+ </summary>
+ </member>
+ <member name="F:NetSharp.ConnectionBuilder.SocketSettings.AddressFamily">
+ <summary>
+ The address family for the socket underlying the connection.
+ </summary>
+ </member>
+ <member name="F:NetSharp.ConnectionBuilder.SocketSettings.ProtocolType">
+ <summary>
+ The socket type for the socket underlying the connection.
+ </summary>
+ </member>
+ <member name="F:NetSharp.ConnectionBuilder.SocketSettings.SocketType">
+ <summary>
+ The protocol type for the socket underlying the connection.
+ </summary>
+ </member>
+ <member name="M:NetSharp.ConnectionBuilder.SocketSettings.#ctor(System.Net.Sockets.AddressFamily,System.Net.Sockets.SocketType,System.Net.Sockets.ProtocolType)">
+ <summary>
+ Initialises a new instance of the <see cref="T:NetSharp.ConnectionBuilder.SocketSettings"/> struct.
+ </summary>
+ <param name="addressFamily">The address family for the underlying socket.</param>
+ <param name="socketType">The socket type for the underlying socket.</param>
+ <param name="protocolType">The protocol type for the underlying socket.</param>
+ </member>
<member name="T:NetSharp.Deprecated.Client">
<summary>
Provides methods for connecting to and talking with a <see cref="T:NetSharp.Deprecated.IServer"/> instance.
@@ -1133,6 +1295,11 @@
<member name="P:NetSharp.Deprecated.UdpSocketOptions.UseLoopback">
<inheritdoc />
</member>
+ <member name="T:NetSharp.Extensions.ConnectionBuilderExtensions">
+ <summary>
+ Provides additional methods and functionality to the <see cref="T:NetSharp.ConnectionBuilder"/> class.
+ </summary>
+ </member>
<member name="T:NetSharp.Extensions.ConnectionExtensions">
<summary>
Provides additional methods and functionality to the <see cref="T:NetSharp.Connection"/> class.
@@ -1727,14 +1894,73 @@
The custom type id that the decorated packet type should have. This overrides the automatically generated id.
</summary>
</member>
- <member name="P:NetSharp.Pipelines.PacketPipelineStage`2.StageInput">
- <inheritdoc />
+ <member name="T:NetSharp.Pipelines.PacketPipeline`3">
+ <summary>
+ Represents a pipeline of transformations that packets must undergo.
+ </summary>
+ <typeparam name="TInput">The type of packet the pipeline receives.</typeparam>
+ <typeparam name="TIntermediate">The type of packet the pipeline internally handles.</typeparam>
+ <typeparam name="TOutput">The type of packet the pipeline outputs.</typeparam>
</member>
- <member name="P:NetSharp.Pipelines.PacketPipelineStage`2.StageOutput">
- <inheritdoc />
+ <member name="M:NetSharp.Pipelines.PacketPipeline`3.ProcessPacket(`0)">
+ <summary>
+ Passes the given packet through the pipeline.
+ </summary>
+ <param name="inputPacket">The incoming packet.</param>
+ <returns>The outgoing transformed packet.</returns>
</member>
- <member name="M:NetSharp.Pipelines.PacketPipelineStage`2.Process(`0)">
- <inheritdoc />
+ <member name="T:NetSharp.Pipelines.PacketPipelineStage`2">
+ <summary>
+ Represents a single transformation applied to a packet traveling through the pipeline.
+ </summary>
+ <typeparam name="TInput">The type the transformation takes as input.</typeparam>
+ <typeparam name="TOutput">The type the transformation produces as output.</typeparam>
+ </member>
+ <member name="T:NetSharp.Pipelines.PacketPipelineBuilder`3">
+ <summary>
+ Allows for configuring and subsequently building a <see cref="T:NetSharp.Pipelines.PacketPipeline`3"/> instance.
+ </summary>
+ <typeparam name="TInput">The type of packet that will be submitted to the pipeline.</typeparam>
+ <typeparam name="TIntermediate">The type of packet that will be handled internally by the pipeline.</typeparam>
+ <typeparam name="TOutput">The type of packet that will be requested from the pipeline.</typeparam>
+ </member>
+ <member name="M:NetSharp.Pipelines.PacketPipelineBuilder`3.Build">
+ <summary>
+ Returns the currently configured <see cref="T:NetSharp.Pipelines.PacketPipeline`3"/> instance.
+ </summary>
+ <returns>The configured <see cref="T:NetSharp.Pipelines.PacketPipeline`3"/> instance.</returns>
+ <exception cref="T:System.ArgumentNullException">
+ Thrown when either <see cref="M:NetSharp.Pipelines.PacketPipelineBuilder`3.WithInputStage(System.Func{`0,`1}@)"/> or <see cref="M:NetSharp.Pipelines.PacketPipelineBuilder`3.WithOutputStage(System.Func{`1,`2}@)"/> have not been called.
+ </exception>
+ </member>
+ <member name="M:NetSharp.Pipelines.PacketPipelineBuilder`3.WithInputStage(System.Func{`0,`1}@)">
+ <summary>
+ Configures the input stage for the pipeline.
+ </summary>
+ <param name="stage">
+ The transformation that should be applied to incoming packets, to convert them from the <typeparamref name="TInput"/>
+ type to the <typeparamref name="TIntermediate"/> type that the pipeline handles internally.
+ </param>
+ <returns>The builder instance for further configuration.</returns>
+ </member>
+ <member name="M:NetSharp.Pipelines.PacketPipelineBuilder`3.WithIntermediateStage(System.Func{`1,`1}@)">
+ <summary>
+ Adds the given intermediate stage to the pipeline.
+ </summary>
+ <param name="stage">
+ The transformation that should be applied to packets traveling through the pipeline.
+ </param>
+ <returns>The builder instance for further configuration.</returns>
+ </member>
+ <member name="M:NetSharp.Pipelines.PacketPipelineBuilder`3.WithOutputStage(System.Func{`1,`2}@)">
+ <summary>
+ Configures the output stage for the pipeline.
+ </summary>
+ <param name="stage">
+ The transformation that should be applied to outgoing packets, to convert them from the
+ <typeparamref name="TIntermediate"/> type used internally to the <typeparamref name="TOutput"/> type.
+ </param>
+ <returns>The builder instance for further configuration.</returns>
</member>
<member name="T:NetSharp.Sockets.SocketAcceptor">
<summary>
diff --git a/NetSharp/NetSharp/Packets/NetworkPacket.cs b/NetSharp/NetSharp/Packets/NetworkPacket.cs
@@ -41,7 +41,7 @@ namespace NetSharp.Packets
/// <summary>
/// The size of each packet, including its header, footer, and data segment.
/// </summary>
- public const int PacketSize = 1024;
+ public const int PacketSize = 4096;
/// <summary>
/// The data held in this packet.
diff --git a/NetSharp/NetSharp/Pipelines/PacketPipeline.cs b/NetSharp/NetSharp/Pipelines/PacketPipeline.cs
@@ -1,84 +1,65 @@
using System;
using System.Collections.Generic;
-using System.Threading;
-using System.Threading.Channels;
-using System.Threading.Tasks;
-using NetSharp.Packets;
+using System.Linq;
namespace NetSharp.Pipelines
{
- public interface IPacketPipelineStage<TInput, TOutput>
- {
- ChannelReader<TInput> StageInput { get; }
-
- ChannelWriter<TOutput> StageOutput { get; }
-
- TOutput Process(TInput input);
- }
-
+ /// <summary>
+ /// Represents a pipeline of transformations that packets must undergo.
+ /// </summary>
+ /// <typeparam name="TInput">The type of packet the pipeline receives.</typeparam>
+ /// <typeparam name="TIntermediate">The type of packet the pipeline internally handles.</typeparam>
+ /// <typeparam name="TOutput">The type of packet the pipeline outputs.</typeparam>
// TODO: Implement a packet pipeline, with multiple transform stages to allow encryption, compression, and various other bytewise manipulation stages.
- public readonly struct PacketPipeline<TInput, TIntermediate, TOutput>
+ internal readonly struct PacketPipeline<TInput, TIntermediate, TOutput>
{
- private readonly IPacketPipelineStage<TIntermediate, TOutput> finalPipelineStage;
- private readonly IPacketPipelineStage<TInput, TIntermediate> initialPipelineStage;
- private readonly IEnumerable<IPacketPipelineStage<TIntermediate, TIntermediate>> intermediatePipelineStages;
- private readonly Channel<TInput> pipelineInput;
- private readonly Channel<TOutput> pipelineOutput;
- private readonly CancellationToken pipelineShutdownToken;
+ private readonly PacketPipelineStage<TInput, TIntermediate> pipelineInputStage;
+ private readonly IReadOnlyCollection<PacketPipelineStage<TIntermediate, TIntermediate>> pipelineIntermediateStages;
+ private readonly PacketPipelineStage<TIntermediate, TOutput> pipelineOutputStage;
- internal PacketPipeline(CancellationToken shutdownToken,
- IPacketPipelineStage<TInput, TIntermediate> firstStage,
- IPacketPipelineStage<TIntermediate, TOutput> lastStage,
- IEnumerable<IPacketPipelineStage<TIntermediate, TIntermediate>> intermediateStages)
+ internal PacketPipeline(
+ PacketPipelineStage<TInput, TIntermediate> firstStage,
+ PacketPipelineStage<TIntermediate, TOutput> lastStage,
+ IReadOnlyCollection<PacketPipelineStage<TIntermediate, TIntermediate>> intermediateStages)
{
- pipelineShutdownToken = shutdownToken;
-
- UnboundedChannelOptions inputOptions = new UnboundedChannelOptions
- {
- SingleReader = true,
- SingleWriter = true,
- };
- pipelineInput = Channel.CreateUnbounded<TInput>(inputOptions);
-
- UnboundedChannelOptions outputOptions = new UnboundedChannelOptions
- {
- SingleReader = true,
- SingleWriter = true,
- };
- pipelineOutput = Channel.CreateUnbounded<TOutput>(outputOptions);
+ pipelineInputStage = firstStage;
+ pipelineOutputStage = lastStage;
- initialPipelineStage = firstStage;
- finalPipelineStage = lastStage;
-
- intermediatePipelineStages = intermediateStages;
- foreach (IPacketPipelineStage<TIntermediate, TIntermediate> stage in intermediatePipelineStages)
- {
- }
+ pipelineIntermediateStages = intermediateStages;
}
- public async Task<TOutput> DequeuePacketAsync()
+ /// <summary>
+ /// Passes the given packet through the pipeline.
+ /// </summary>
+ /// <param name="inputPacket">The incoming packet.</param>
+ /// <returns>The outgoing transformed packet.</returns>
+ internal TOutput ProcessPacket(TInput inputPacket)
{
- return await pipelineOutput.Reader.ReadAsync(pipelineShutdownToken);
- }
+ TIntermediate intermediatePacket = pipelineInputStage.Process(inputPacket);
- public async Task EnqueuePacketAsync(TInput pipelineInput)
- {
- await this.pipelineInput.Writer.WriteAsync(pipelineInput, pipelineShutdownToken);
+ intermediatePacket = pipelineIntermediateStages.Aggregate(intermediatePacket, (current, stage) => stage.Process(current));
+
+ return pipelineOutputStage.Process(intermediatePacket);
}
}
- public readonly struct PacketPipelineStage<TInput, TOutput> : IPacketPipelineStage<TInput, TOutput>
+ /// <summary>
+ /// Represents a single transformation applied to a packet traveling through the pipeline.
+ /// </summary>
+ /// <typeparam name="TInput">The type the transformation takes as input.</typeparam>
+ /// <typeparam name="TOutput">The type the transformation produces as output.</typeparam>
+ internal readonly struct PacketPipelineStage<TInput, TOutput>
{
- /// <inheritdoc />
- public ChannelReader<TInput> StageInput { get; }
+ private readonly Func<TInput, TOutput> stageDelegate;
- /// <inheritdoc />
- public ChannelWriter<TOutput> StageOutput { get; }
+ internal PacketPipelineStage(in Func<TInput, TOutput> stageProcessingDelegate)
+ {
+ stageDelegate = stageProcessingDelegate;
+ }
- /// <inheritdoc />
- public TOutput Process(TInput input)
+ internal TOutput Process(TInput input)
{
- throw new NotImplementedException();
+ return stageDelegate(input);
}
}
}
\ No newline at end of file
diff --git a/NetSharp/NetSharp/Pipelines/PacketPipelineBuilder.cs b/NetSharp/NetSharp/Pipelines/PacketPipelineBuilder.cs
@@ -0,0 +1,87 @@
+using System;
+using System.Collections.Generic;
+
+namespace NetSharp.Pipelines
+{
+ /// <summary>
+ /// Allows for configuring and subsequently building a <see cref="PacketPipeline{TInput,TIntermediate,TOutput}"/> instance.
+ /// </summary>
+ /// <typeparam name="TInput">The type of packet that will be submitted to the pipeline.</typeparam>
+ /// <typeparam name="TIntermediate">The type of packet that will be handled internally by the pipeline.</typeparam>
+ /// <typeparam name="TOutput">The type of packet that will be requested from the pipeline.</typeparam>
+ internal sealed class PacketPipelineBuilder<TInput, TIntermediate, TOutput>
+ {
+ private readonly List<PacketPipelineStage<TIntermediate, TIntermediate>> intermediateStages;
+
+ private PacketPipelineStage<TInput, TIntermediate>? inputStage;
+ private PacketPipelineStage<TIntermediate, TOutput>? outputStage;
+
+ internal PacketPipelineBuilder()
+ {
+ intermediateStages = new List<PacketPipelineStage<TIntermediate, TIntermediate>>();
+ }
+
+ /// <summary>
+ /// Returns the currently configured <see cref="PacketPipeline{TInput,TIntermediate,TOutput}"/> instance.
+ /// </summary>
+ /// <returns>The configured <see cref="PacketPipeline{TInput,TIntermediate,TOutput}"/> instance.</returns>
+ /// <exception cref="ArgumentNullException">
+ /// Thrown when either <see cref="WithInputStage"/> or <see cref="WithOutputStage"/> have not been called.
+ /// </exception>
+ internal PacketPipeline<TInput, TIntermediate, TOutput> Build()
+ {
+ if (inputStage == null)
+ {
+ throw new ArgumentNullException(nameof(inputStage), $"{nameof(WithInputStage)} has not been called.");
+ }
+
+ if (outputStage == null)
+ {
+ throw new ArgumentNullException(nameof(outputStage), $"{nameof(WithOutputStage)} has not been called.");
+ }
+
+ return new PacketPipeline<TInput, TIntermediate, TOutput>(inputStage.Value, outputStage.Value, intermediateStages);
+ }
+
+ /// <summary>
+ /// Configures the input stage for the pipeline.
+ /// </summary>
+ /// <param name="stage">
+ /// The transformation that should be applied to incoming packets, to convert them from the <typeparamref name="TInput"/>
+ /// type to the <typeparamref name="TIntermediate"/> type that the pipeline handles internally.
+ /// </param>
+ /// <returns>The builder instance for further configuration.</returns>
+ internal PacketPipelineBuilder<TInput, TIntermediate, TOutput> WithInputStage(in Func<TInput, TIntermediate> stage)
+ {
+ inputStage = new PacketPipelineStage<TInput, TIntermediate>(in stage);
+ return this;
+ }
+
+ /// <summary>
+ /// Adds the given intermediate stage to the pipeline.
+ /// </summary>
+ /// <param name="stage">
+ /// The transformation that should be applied to packets traveling through the pipeline.
+ /// </param>
+ /// <returns>The builder instance for further configuration.</returns>
+ internal PacketPipelineBuilder<TInput, TIntermediate, TOutput> WithIntermediateStage(in Func<TIntermediate, TIntermediate> stage)
+ {
+ intermediateStages.Add(new PacketPipelineStage<TIntermediate, TIntermediate>(in stage));
+ return this;
+ }
+
+ /// <summary>
+ /// Configures the output stage for the pipeline.
+ /// </summary>
+ /// <param name="stage">
+ /// The transformation that should be applied to outgoing packets, to convert them from the
+ /// <typeparamref name="TIntermediate"/> type used internally to the <typeparamref name="TOutput"/> type.
+ /// </param>
+ /// <returns>The builder instance for further configuration.</returns>
+ internal PacketPipelineBuilder<TInput, TIntermediate, TOutput> WithOutputStage(in Func<TIntermediate, TOutput> stage)
+ {
+ outputStage = new PacketPipelineStage<TIntermediate, TOutput>(in stage);
+ return this;
+ }
+ }
+}
+\ No newline at end of file
diff --git a/NetSharp/NetSharp/Sockets/SocketAcceptor.cs b/NetSharp/NetSharp/Sockets/SocketAcceptor.cs
@@ -177,6 +177,7 @@ namespace NetSharp.Sockets
SocketAsyncEventArgs args = acceptAsyncEventArgsPool.Get();
args.UserToken = new AsyncAcceptToken(tcs, cancellationToken);
+ /*
// register cleanup action for when the cancellation token is thrown
cancellationToken.Register(() =>
{
@@ -190,6 +191,7 @@ namespace NetSharp.Sockets
newArgs.Completed += HandleIOCompleted;
acceptAsyncEventArgsPool.Return(newArgs);
});
+ */
// if the accept operation doesn't complete synchronously, return the awaitable task
if (socket.AcceptAsync(args)) return tcs.Task;
@@ -215,6 +217,7 @@ namespace NetSharp.Sockets
args.RemoteEndPoint = remoteEndPoint;
args.UserToken = new AsyncConnectToken(tcs, cancellationToken);
+ /*
// register cleanup action for when the cancellation token is thrown
cancellationToken.Register(() =>
{
@@ -228,6 +231,7 @@ namespace NetSharp.Sockets
newArgs.Completed += HandleIOCompleted;
connectAsyncEventArgsPool.Return(newArgs);
});
+ */
// if the connect operation doesn't complete synchronously, return the awaitable task
if (socket.ConnectAsync(args)) return tcs.Task;
@@ -249,6 +253,7 @@ namespace NetSharp.Sockets
SocketAsyncEventArgs args = connectAsyncEventArgsPool.Get();
args.UserToken = new AsyncDisconnectToken(tcs, cancellationToken);
+ /*
// register cleanup action for when the cancellation token is thrown
cancellationToken.Register(() =>
{
@@ -262,6 +267,7 @@ namespace NetSharp.Sockets
newArgs.Completed += HandleIOCompleted;
disconnectAsyncEventArgsPool.Return(newArgs);
});
+ */
// if the disconnect operation doesn't complete synchronously, return the awaitable task
if (socket.DisconnectAsync(args)) return tcs.Task;
diff --git a/NetSharp/NetSharp/Sockets/SocketReader.cs b/NetSharp/NetSharp/Sockets/SocketReader.cs
@@ -145,6 +145,7 @@ namespace NetSharp.Sockets
args.SocketFlags = socketFlags;
args.UserToken = new AsyncReadToken(rentedReceiveBuffer, inputBuffer, tcs, cancellationToken);
+ /*
// register cleanup action for when the cancellation token is thrown
cancellationToken.Register(() =>
{
@@ -160,6 +161,7 @@ namespace NetSharp.Sockets
newArgs.Completed += HandleIOCompleted;
receiveAsyncEventArgsPool.Return(newArgs);
});
+ */
// if the receive operation doesn't complete synchronously, returns the awaitable task
if (socket.ReceiveAsync(args)) return tcs.Task;
@@ -197,6 +199,7 @@ namespace NetSharp.Sockets
args.RemoteEndPoint = remoteEndPoint;
args.UserToken = new AsyncReadToken(rentedReceiveFromBuffer, inputBuffer, tcs, cancellationToken);
+ /*
// register cleanup action for when the cancellation token is thrown
cancellationToken.Register(() =>
{
@@ -212,6 +215,7 @@ namespace NetSharp.Sockets
newArgs.Completed += HandleIOCompleted;
receiveAsyncEventArgsPool.Return(newArgs);
});
+ */
// if the receive operation doesn't complete synchronously, returns the awaitable task
if (socket.ReceiveFromAsync(args)) return tcs.Task;
diff --git a/NetSharp/NetSharp/Sockets/SocketWriter.cs b/NetSharp/NetSharp/Sockets/SocketWriter.cs
@@ -134,6 +134,24 @@ namespace NetSharp.Sockets
args.SocketFlags = socketFlags;
args.UserToken = new AsyncWriteToken(rentedSendBuffer, tcs, cancellationToken);
+ /*
+ // register cleanup action for when the cancellation token is thrown
+ cancellationToken.Register(() =>
+ {
+ tcs.SetCanceled();
+
+ sendBufferPool.Return(rentedSendBuffer, true);
+
+ //TODO this is probably a hideous solution. find a better one
+ args.Completed -= HandleIOCompleted;
+ args.Dispose();
+
+ SocketAsyncEventArgs newArgs = new SocketAsyncEventArgs();
+ newArgs.Completed += HandleIOCompleted;
+ sendAsyncEventArgsPool.Return(newArgs);
+ });
+ */
+
// if the send operation doesn't complete synchronously, return the awaitable task
if (socket.SendAsync(args)) return tcs.Task;
@@ -170,6 +188,7 @@ namespace NetSharp.Sockets
args.RemoteEndPoint = remoteEndPoint;
args.UserToken = new AsyncWriteToken(rentedSendToBuffer, tcs, cancellationToken);
+ /*
// register cleanup action for when the cancellation token is thrown
cancellationToken.Register(() =>
{
@@ -185,6 +204,7 @@ namespace NetSharp.Sockets
newArgs.Completed += HandleIOCompleted;
sendAsyncEventArgsPool.Return(newArgs);
});
+ */
// if the send operation doesn't complete synchronously, return the awaitable task
if (socket.SendToAsync(args)) return tcs.Task;
diff --git a/NetSharp/NetSharpExamples/Program.cs b/NetSharp/NetSharpExamples/Program.cs
@@ -4,8 +4,10 @@ using System.IO;
using System.Net;
using System.Net.Sockets;
using System.Text;
+using System.Threading;
using System.Threading.Tasks;
using NetSharp;
+using NetSharp.Extensions;
using NetSharp.Packets;
using NetSharp.Utils;
@@ -25,19 +27,8 @@ namespace NetSharpExamples
serverAddress = IPAddress.Loopback;
serverPort = 12374;
- Console.WriteLine("Test server (y/n): ");
- if (Console.ReadLine()?.ToLower().Equals("y") ?? false)
- {
- await TestServer();
-
- Console.ReadLine();
- }
- else
- {
- await TestClient();
-
- Console.ReadLine();
- }
+ await Task.Factory.StartNew(TestServer);
+ await Task.Factory.StartNew(TestClient).Result;
}
private static async Task TestClient()
@@ -45,13 +36,14 @@ namespace NetSharpExamples
TimeSpan socketTimeout = TimeSpan.FromSeconds(newtorkTimeout);
const int clientCount = 10;
- const int sentPacketCount = 1_000_000;
+ const long sentPacketCount = 10_000;
EndPoint serverEndPoint = new IPEndPoint(serverAddress, serverPort);
+ ConnectionBuilder clientBuilder = new ConnectionBuilder().WithUdp();
- static Connection ClientFactory()
+ Connection ClientFactory()
{
- return ConnectionFactory.InitUdp(new IPEndPoint(IPAddress.Loopback, 0));
+ return clientBuilder.Build();
}
Console.WriteLine($"Testing client connections...");
@@ -63,7 +55,9 @@ namespace NetSharpExamples
Console.WriteLine($"Starting client {clientId}");
using Connection client = ClientFactory();
- TimeSpan timeout = TimeSpan.FromMilliseconds(1);
+ client.TryBind(new IPEndPoint(IPAddress.Any, 0));
+ //TimeSpan timeout = TimeSpan.FromMilliseconds(100);
+ Stopwatch stopwatch = new Stopwatch();
byte[] message = Encoding.UTF8.GetBytes("Hello World!");
Memory<byte> messageBuffer = new Memory<byte>(message);
@@ -71,18 +65,38 @@ namespace NetSharpExamples
byte[] response = new byte[NetworkPacket.PacketSize];
Memory<byte> responseBuffer = new Memory<byte>(response);
+ long sentPackets = 0, receivedPackets = 0;
for (int j = 0; j < sentPacketCount; j++)
{
- int sentBytes = await client.SendToAsync(serverEndPoint, messageBuffer, SocketFlags.None, timeout);
-
- Console.WriteLine($"[Client {clientId}] Sent {sentBytes} bytes to {serverEndPoint}");
-
- TransmissionResult result = await client.ReceiveFromAsync(serverEndPoint, responseBuffer, SocketFlags.None, timeout);
+ try
+ {
+ stopwatch.Start();
+ int sentBytes = await client.SendToAsync(serverEndPoint, messageBuffer, SocketFlags.None);
+ stopwatch.Stop();
+ Interlocked.Increment(ref sentPackets);
+
+ //Console.WriteLine($"[Client {clientId}] Sent {sentBytes} bytes to {serverEndPoint}");
+
+ stopwatch.Start();
+ TransmissionResult result = await client.ReceiveFromAsync(serverEndPoint, responseBuffer, SocketFlags.None);
+ stopwatch.Stop();
+ Interlocked.Increment(ref receivedPackets);
+
+ //Console.WriteLine($"[Client {clientId}] Received {result.Count} bytes from {result.RemoteEndPoint}");
+
+ //await Task.Delay(10);
+ }
+ catch (Exception ex)
+ {
+ Console.WriteLine($"[Client {clientId}] Exception: {ex}");
+ }
+ }
- Console.WriteLine($"[Client {clientId}] Received {result.Count} bytes from {result.RemoteEndPoint}");
+ long millis = stopwatch.ElapsedMilliseconds;
+ double megabytes = sentPackets * NetworkPacket.DataSegmentSize / 1_000_000.0;
- //await Task.Delay(10);
- }
+ Console.WriteLine($"[Client {clientId}] Sent {sentPacketCount} packets to {serverEndPoint} in {millis} milliseconds");
+ Console.WriteLine($"[Client {clientId}] Approximate bandwidth: {megabytes / (millis / 1000.0):F3} MBps");
}, i, TaskCreationOptions.LongRunning);
}
@@ -96,8 +110,10 @@ namespace NetSharpExamples
await using Stream serverOutputStream = File.OpenWrite(serverLogFile);
EndPoint serverEndPoint = new IPEndPoint(serverAddress, serverPort);
+ ConnectionBuilder serverBuilder = new ConnectionBuilder().WithUdp();
- using Connection server = ConnectionFactory.InitUdp(serverEndPoint);
+ using Connection server = serverBuilder.Build();
+ server.TryBind(serverEndPoint);
//server.SetLoggingStream(Console.OpenStandardOutput(), LogLevel.Info);
//server.ChangeLoggingStream(serverOutputStream, LogLevel.Error);
@@ -113,11 +129,11 @@ namespace NetSharpExamples
TransmissionResult result =
await server.ReceiveFromAsync(nullEndPoint, requestBuffer, SocketFlags.None);
- Console.WriteLine($"[Server] Received {result.Count} bytes from {result.RemoteEndPoint}");
+ //Console.WriteLine($"[Server] Received {result.Count} bytes from {result.RemoteEndPoint}");
int sentBytes = await server.SendToAsync(result.RemoteEndPoint, requestBuffer, SocketFlags.None);
- Console.WriteLine($"[Server] Sent {sentBytes} bytes tp {result.RemoteEndPoint}");
+ //Console.WriteLine($"[Server] Sent {sentBytes} bytes tp {result.RemoteEndPoint}");
}
Console.WriteLine("Server stopped");
