neuecc

try
{
    await client.SendAsync(token);
}
catch (OperationCanceledException ex) when (ex.CancellationToken == token)
{
    // detect reason of Cancel by Token
}

neuecc

class Client
{
    public TimeSpan Timeout { get; }

    public Client(TimeSpan timeout)
    {
        this.Timeout = timeout;
    }

    public async Task SendAsync(CancellationToken cancellationToken = default)

neuecc

// Dispose will stop internal timer to prevent timer leak
using var cts = new CancellationTokenSource();
cts.CancelAfter(TimeSpan.FromMinutes(1));

await client.SendAsync(cts.Token);

neuecc

class Client
{
    public async Task SendAsync(CancellationToken cancellationToken = default)
    {
        await SendCoreAsync(cancellationToken);
    }

    async Task SendCoreAsync(CancellationToken cancellationToken)
    {
        // do anything...

neuecc

public interface INatsSerializer
{
    int Serialize<T>(ICountableBufferWriter bufferWriter, T? value);
    T? Deserialize<T>(in ReadOnlySequence<byte> buffer);
}

public interface ICountableBufferWriter : IBufferWriter<byte>
{
    int WrittenCount { get; }
}

neuecc

class AsyncPublishCommand<T> : 
    ICommand,
    IValueTaskSource, 
    IThreadPoolWorkItem, 
    IObjectPoolNode<AsyncPublishCommand<T>>

internal interface ICommand
{
    void Write(ProtocolWriter writer);
}

neuecc

await connection.PublishAsync(value);

neuecc

internal static class ServerOpCodes
{
    public const int Info = 1330007625;  // Encoding.ASCII.GetBytes("INFO") |> MemoryMarshal.Read<int>
    public const int Msg = 541545293;    // Encoding.ASCII.GetBytes("MSG ") |> MemoryMarshal.Read<int>
    public const int Ping = 1196312912;  // Encoding.ASCII.GetBytes("PING") |> MemoryMarshal.Read<int>
    public const int Pong = 1196314448;  // Encoding.ASCII.GetBytes("PONG") |> MemoryMarshal.Read<int>
    public const int Ok = 223039275;     // Encoding.ASCII.GetBytes("+OK\r") |> MemoryMarshal.Read<int>
    public const int Error = 1381123373; // Encoding.ASCII.GetBytes("-ERR") |> MemoryMarshal.Read<int>
}

neuecc

// msg = ReadOnlySpan<byte>
if(Unsafe.ReadUnaligned<int>(ref MemoryMarshal.GetReference<byte>(msg)) == 1330007625) // INFO
{
}

neuecc

// use this
public ValueTask ConnectAsync(string host, int port, CancellationToken cancellationToken)
public ValueTask<int> ReceiveAsync(Memory<byte> buffer, SocketFlags socketFlags, CancellationToken cancellationToken)
public ValueTask<int> SendAsync(ReadOnlyMemory<byte> buffer, SocketFlags socketFlags, CancellationToken cancellationToken))

// don't use this
public Task ConnectAsync(string host, int port)
public Task<int> ReceiveAsync(ArraySegment<byte> buffer, SocketFlags socketFlags)
public Task<int> SendAsync(ArraySegment<byte> buffer, SocketFlags socketFlags)