mxaccessgw/src/MxGateway.Worker/Ipc/WorkerPipeClient.cs

using System;
using System.IO;
using System.IO.Pipes;
using System.Threading;
using System.Threading.Tasks;
using MxGateway.Worker.Bootstrap;

namespace MxGateway.Worker.Ipc;

public sealed class WorkerPipeClient : IWorkerPipeClient
{
    public const int DefaultConnectTimeoutMilliseconds = 30000;

    private readonly int _connectTimeoutMilliseconds;
    private readonly Func<Stream, WorkerFrameProtocolOptions, IWorkerLogger?, WorkerPipeSession> _sessionFactory;
    private readonly IWorkerLogger? _logger;

    public WorkerPipeClient()
        : this(null, DefaultConnectTimeoutMilliseconds)
    {
    }

    public WorkerPipeClient(IWorkerLogger? logger)
        : this(logger, DefaultConnectTimeoutMilliseconds)
    {
    }

    public WorkerPipeClient(int connectTimeoutMilliseconds)
        : this(null, connectTimeoutMilliseconds)
    {
    }

    public WorkerPipeClient(
        int connectTimeoutMilliseconds,
        Func<Stream, WorkerFrameProtocolOptions, WorkerPipeSession> sessionFactory)
        : this(
            null,
            connectTimeoutMilliseconds,
            (stream, frameOptions, _) => sessionFactory(stream, frameOptions))
    {
    }

    public WorkerPipeClient(
        IWorkerLogger? logger,
        int connectTimeoutMilliseconds)
        : this(
            logger,
            connectTimeoutMilliseconds,
            (stream, frameOptions, workerLogger) => new WorkerPipeSession(stream, frameOptions, workerLogger))
    {
    }

    public WorkerPipeClient(
        IWorkerLogger? logger,
        int connectTimeoutMilliseconds,
        Func<Stream, WorkerFrameProtocolOptions, IWorkerLogger?, WorkerPipeSession> sessionFactory)
    {
        if (connectTimeoutMilliseconds <= 0)
        {
            throw new ArgumentOutOfRangeException(
                nameof(connectTimeoutMilliseconds),
                "Worker pipe connect timeout must be greater than zero.");
        }

        _logger = logger;
        _sessionFactory = sessionFactory ?? throw new ArgumentNullException(nameof(sessionFactory));
        _connectTimeoutMilliseconds = connectTimeoutMilliseconds;
    }

    public async Task RunAsync(
        WorkerOptions options,
        CancellationToken cancellationToken = default)
    {
        if (options is null)
        {
            throw new ArgumentNullException(nameof(options));
        }

        WorkerFrameProtocolOptions frameOptions = new(options);

        using NamedPipeClientStream pipe = new(
            ".",
            options.PipeName,
            PipeDirection.InOut,
            PipeOptions.Asynchronous);

        await ConnectAsync(pipe, cancellationToken).ConfigureAwait(false);

        WorkerPipeSession session = _sessionFactory(pipe, frameOptions, _logger);
        await session.RunAsync(cancellationToken).ConfigureAwait(false);
    }

    private Task ConnectAsync(
        NamedPipeClientStream pipe,
        CancellationToken cancellationToken)
    {
        return Task.Run(
            () =>
            {
                cancellationToken.ThrowIfCancellationRequested();
                pipe.Connect(_connectTimeoutMilliseconds);
            },
            cancellationToken);
    }
}