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

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

namespace ZB.MOM.WW.MxGateway.Client;

/// <summary>
/// gRPC implementation of IGalaxyRepositoryClientTransport.
/// </summary>
internal sealed class GrpcGalaxyRepositoryClientTransport(
    MxGatewayClientOptions options,
    GalaxyRepository.GalaxyRepositoryClient rawClient) : IGalaxyRepositoryClientTransport
{
    /// <inheritdoc />
    public MxGatewayClientOptions Options { get; } = options;

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

    /// <inheritdoc />
    GalaxyRepository.GalaxyRepositoryClient? IGalaxyRepositoryClientTransport.RawClient => RawClient;

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

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

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

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

    /// <summary>Streams deploy events from the Galaxy Repository, using an explicit cancellation token that overrides the call options token when provided.</summary>
    /// <param name="request">The watch deploy events request.</param>
    /// <param name="callOptions">Call options for the underlying gRPC call.</param>
    /// <param name="cancellationToken">Optional cancellation token; takes precedence over the token in <paramref name="callOptions"/> when cancellable.</param>
    /// <returns>An async enumerable of deploy events.</returns>
    public async IAsyncEnumerable<DeployEvent> WatchDeployEventsAsync(
        WatchDeployEventsRequest request,
        CallOptions callOptions,
        [System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken cancellationToken = default)
    {
        CancellationToken effectiveCancellationToken = cancellationToken.CanBeCanceled
            ? cancellationToken
            : callOptions.CancellationToken;

        using AsyncServerStreamingCall<DeployEvent> call = RawClient.WatchDeployEvents(request, callOptions);

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

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

            yield return deployEvent;
        }
    }

    /// <inheritdoc />
    IAsyncEnumerable<DeployEvent> IGalaxyRepositoryClientTransport.WatchDeployEventsAsync(
        WatchDeployEventsRequest request,
        CallOptions callOptions)
    {
        return WatchDeployEventsAsync(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),
        };
    }
}