mxaccessgw/clients/dotnet/ZB.MOM.WW.MxGateway.Client/GrpcMxGatewayClientTransport.cs

using Grpc.Core;
using ZB.MOM.WW.MxGateway.Contracts.Proto;

namespace ZB.MOM.WW.MxGateway.Client;

/// <summary>
/// gRPC implementation of IMxGatewayClientTransport.
/// </summary>
internal sealed class GrpcMxGatewayClientTransport(
    MxGatewayClientOptions options,
    MxAccessGateway.MxAccessGatewayClient rawClient) : IMxGatewayClientTransport
{
    /// <inheritdoc />
    public MxGatewayClientOptions Options { get; } = options;

    /// <summary>
    /// Gets the underlying gRPC client.
    /// </summary>
    public MxAccessGateway.MxAccessGatewayClient RawClient { get; } = rawClient;

    /// <inheritdoc />
    MxAccessGateway.MxAccessGatewayClient? IMxGatewayClientTransport.RawClient => RawClient;

    /// <inheritdoc />
    public async Task<OpenSessionReply> OpenSessionAsync(
        OpenSessionRequest request,
        CallOptions callOptions)
    {
        try
        {
            return await RawClient.OpenSessionAsync(request, callOptions)
                .ResponseAsync
                .ConfigureAwait(false);
        }
        catch (RpcException exception)
        {
            throw MapRpcException(exception, callOptions.CancellationToken);
        }
    }

    /// <inheritdoc />
    public async Task<CloseSessionReply> CloseSessionAsync(
        CloseSessionRequest request,
        CallOptions callOptions)
    {
        try
        {
            return await RawClient.CloseSessionAsync(request, callOptions)
                .ResponseAsync
                .ConfigureAwait(false);
        }
        catch (RpcException exception)
        {
            throw MapRpcException(exception, callOptions.CancellationToken);
        }
    }

    /// <inheritdoc />
    public async Task<MxCommandReply> InvokeAsync(
        MxCommandRequest request,
        CallOptions callOptions)
    {
        try
        {
            return await RawClient.InvokeAsync(request, callOptions)
                .ResponseAsync
                .ConfigureAwait(false);
        }
        catch (RpcException exception)
        {
            throw MapRpcException(exception, callOptions.CancellationToken);
        }
    }

    /// <summary>
    /// Streams events from the session.
    /// </summary>
    /// <param name="request">The stream events request.</param>
    /// <param name="callOptions">gRPC call options.</param>
    /// <param name="cancellationToken">Cancellation token; falls back to <paramref name="callOptions"/>'s token when not cancelable.</param>
    /// <returns>An async enumerable of events.</returns>
    public async IAsyncEnumerable<MxEvent> StreamEventsAsync(
        StreamEventsRequest request,
        CallOptions callOptions,
        [System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken cancellationToken = default)
    {
        CancellationToken effectiveCancellationToken = cancellationToken.CanBeCanceled
            ? cancellationToken
            : callOptions.CancellationToken;

        using AsyncServerStreamingCall<MxEvent> call = RawClient.StreamEvents(request, callOptions);

        IAsyncStreamReader<MxEvent> responseStream = call.ResponseStream;
        while (true)
        {
            MxEvent? gatewayEvent;
            try
            {
                if (!await responseStream.MoveNext(effectiveCancellationToken).ConfigureAwait(false))
                {
                    break;
                }

                gatewayEvent = responseStream.Current;
            }
            catch (RpcException exception)
            {
                throw MapRpcException(exception, effectiveCancellationToken);
            }

            yield return gatewayEvent;
        }
    }

    /// <inheritdoc />
    IAsyncEnumerable<MxEvent> IMxGatewayClientTransport.StreamEventsAsync(
        StreamEventsRequest request,
        CallOptions callOptions)
    {
        return StreamEventsAsync(request, callOptions);
    }

    /// <inheritdoc />
    public async Task<AcknowledgeAlarmReply> AcknowledgeAlarmAsync(
        AcknowledgeAlarmRequest request,
        CallOptions callOptions)
    {
        try
        {
            return await RawClient.AcknowledgeAlarmAsync(request, callOptions)
                .ResponseAsync
                .ConfigureAwait(false);
        }
        catch (RpcException exception)
        {
            throw MapRpcException(exception, callOptions.CancellationToken);
        }
    }

    /// <summary>
    /// Streams a snapshot of all alarms currently in Active or ActiveAcked state — the
    /// ConditionRefresh equivalent for the gateway.
    /// </summary>
    /// <param name="request">The query request, optionally scoped by alarm-reference prefix.</param>
    /// <param name="callOptions">gRPC call options.</param>
    /// <param name="cancellationToken">Cancellation token; falls back to <paramref name="callOptions"/>'s token when not cancelable.</param>
    /// <returns>An async enumerable of active-alarm snapshots.</returns>
    public async IAsyncEnumerable<ActiveAlarmSnapshot> QueryActiveAlarmsAsync(
        QueryActiveAlarmsRequest request,
        CallOptions callOptions,
        [System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken cancellationToken = default)
    {
        CancellationToken effectiveCancellationToken = cancellationToken.CanBeCanceled
            ? cancellationToken
            : callOptions.CancellationToken;

        using AsyncServerStreamingCall<ActiveAlarmSnapshot> call = RawClient.QueryActiveAlarms(request, callOptions);

        IAsyncStreamReader<ActiveAlarmSnapshot> responseStream = call.ResponseStream;
        while (true)
        {
            ActiveAlarmSnapshot? snapshot;
            try
            {
                if (!await responseStream.MoveNext(effectiveCancellationToken).ConfigureAwait(false))
                {
                    break;
                }

                snapshot = responseStream.Current;
            }
            catch (RpcException exception)
            {
                throw MapRpcException(exception, effectiveCancellationToken);
            }

            yield return snapshot;
        }
    }

    /// <inheritdoc />
    IAsyncEnumerable<ActiveAlarmSnapshot> IMxGatewayClientTransport.QueryActiveAlarmsAsync(
        QueryActiveAlarmsRequest request,
        CallOptions callOptions)
    {
        return QueryActiveAlarmsAsync(request, callOptions);
    }

    /// <summary>
    /// Attaches to the gateway's central alarm feed — the current active-alarm
    /// snapshot followed by live transitions.
    /// </summary>
    /// <param name="request">The stream request, optionally scoped by alarm-reference prefix.</param>
    /// <param name="callOptions">gRPC call options.</param>
    /// <param name="cancellationToken">Cancellation token; falls back to <paramref name="callOptions"/>'s token when not cancelable.</param>
    /// <returns>An async enumerable of alarm feed messages.</returns>
    public async IAsyncEnumerable<AlarmFeedMessage> StreamAlarmsAsync(
        StreamAlarmsRequest request,
        CallOptions callOptions,
        [System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken cancellationToken = default)
    {
        CancellationToken effectiveCancellationToken = cancellationToken.CanBeCanceled
            ? cancellationToken
            : callOptions.CancellationToken;

        using AsyncServerStreamingCall<AlarmFeedMessage> call = RawClient.StreamAlarms(request, callOptions);

        IAsyncStreamReader<AlarmFeedMessage> responseStream = call.ResponseStream;
        while (true)
        {
            AlarmFeedMessage? message;
            try
            {
                if (!await responseStream.MoveNext(effectiveCancellationToken).ConfigureAwait(false))
                {
                    break;
                }

                message = responseStream.Current;
            }
            catch (RpcException exception)
            {
                throw MapRpcException(exception, effectiveCancellationToken);
            }

            yield return message;
        }
    }

    /// <inheritdoc />
    IAsyncEnumerable<AlarmFeedMessage> IMxGatewayClientTransport.StreamAlarmsAsync(
        StreamAlarmsRequest request,
        CallOptions callOptions)
    {
        return StreamAlarmsAsync(request, callOptions);
    }

    private static Exception MapRpcException(
        RpcException exception,
        CancellationToken cancellationToken)
    {
        if (cancellationToken.IsCancellationRequested || exception.StatusCode == StatusCode.Cancelled)
        {
            return new OperationCanceledException(
                exception.Status.Detail,
                exception,
                cancellationToken);
        }

        return exception.StatusCode switch
        {
            StatusCode.Unauthenticated => new MxGatewayAuthenticationException(
                exception.Status.Detail,
                innerException: exception),
            StatusCode.PermissionDenied => new MxGatewayAuthorizationException(
                exception.Status.Detail,
                innerException: exception),
            _ => new MxGatewayException(exception.Status.Detail, exception),
        };
    }
}