src/MassTransit/Transports/SendEndpointProxy.cs - develop - vshmidt/masstransit

src/MassTransit/Transports/SendEndpointProxy.cs
160 строк6 KB
Chris Patterson
Cleaned up leftover container abstractions that were no longer useful
04 дек 2022, 21:03
04 дек 2022, 21:0368eb905
#nullable enable
namespace MassTransit.Transports
{
    using System;
    using System.Threading;
    using System.Threading.Tasks;
    using Context;
    using Initializers;


    /// <summary>
    /// Generalized proxy for ISendEndpoint to intercept pipe/context
    /// </summary>
    public abstract class SendEndpointProxy :
        ITransportSendEndpoint
    {
        readonly ISendEndpoint _endpoint;
        readonly ITransportSendEndpoint? _transportEndpoint;

        protected SendEndpointProxy(ISendEndpoint endpoint)
        {
            _endpoint = endpoint;
            _transportEndpoint = endpoint as ITransportSendEndpoint;
        }

        public ISendEndpoint Endpoint => _endpoint;

        public ConnectHandle ConnectSendObserver(ISendObserver observer)
        {
            return _endpoint.ConnectSendObserver(observer);
        }

        public Task<SendContext<T>> CreateSendContext<T>(T message, IPipe<SendContext<T>> pipe, CancellationToken cancellationToken)
            where T : class
        {
            return _transportEndpoint?.CreateSendContext(message, GetPipeProxy(pipe), cancellationToken)
                ?? throw new InvalidOperationException("The endpoint does not have a valid transport");
        }

        public virtual Task Send<T>(T message, CancellationToken cancellationToken)
            where T : class
        {
            if (message == null)
                throw new ArgumentNullException(nameof(message));

            return _endpoint.Send(message, GetPipeProxy<T>(), cancellationToken);
        }

        public virtual Task Send<T>(T message, IPipe<SendContext<T>> pipe, CancellationToken cancellationToken)
            where T : class
        {
            if (message == null)
                throw new ArgumentNullException(nameof(message));

            if (pipe == null)
                throw new ArgumentNullException(nameof(pipe));

            return _endpoint.Send(message, GetPipeProxy(pipe), cancellationToken);
        }

        public virtual Task Send<T>(T message, IPipe<SendContext> pipe, CancellationToken cancellationToken)
            where T : class
        {
            if (message == null)
                throw new ArgumentNullException(nameof(message));

            if (pipe == null)
                throw new ArgumentNullException(nameof(pipe));

            return _endpoint.Send(message, GetPipeProxy<T>(pipe), cancellationToken);
        }

        public Task Send(object message, CancellationToken cancellationToken)
        {
            if (message == null)
                throw new ArgumentNullException(nameof(message));

            var messageType = message.GetType();

            return SendEndpointConverterCache.Send(this, message, messageType, cancellationToken);
        }

        public Task Send(object message, Type messageType, CancellationToken cancellationToken)
        {
            if (message == null)
                throw new ArgumentNullException(nameof(message));
            if (messageType == null)
                throw new ArgumentNullException(nameof(messageType));

            return SendEndpointConverterCache.Send(this, message, messageType, cancellationToken);
        }

        public Task Send(object message, IPipe<SendContext> pipe, CancellationToken cancellationToken)
        {
            if (message == null)
                throw new ArgumentNullException(nameof(message));
            if (pipe == null)
                throw new ArgumentNullException(nameof(pipe));

            var messageType = message.GetType();

            return SendEndpointConverterCache.Send(this, message, messageType, pipe, cancellationToken);
        }

        public Task Send(object message, Type messageType, IPipe<SendContext> pipe, CancellationToken cancellationToken)
        {
            if (message == null)
                throw new ArgumentNullException(nameof(message));
            if (messageType == null)
                throw new ArgumentNullException(nameof(messageType));
            if (pipe == null)
                throw new ArgumentNullException(nameof(pipe));

            return SendEndpointConverterCache.Send(this, message, messageType, pipe, cancellationToken);
        }

        public async Task Send<T>(object values, CancellationToken cancellationToken)
            where T : class
        {
            if (values == null)
                throw new ArgumentNullException(nameof(values));

            (var message, IPipe<SendContext<T>> sendPipe) =
                await MessageInitializerCache<T>.InitializeMessage(values, GetPipeProxy<T>(), cancellationToken).ConfigureAwait(false);

            await _endpoint.Send(message, sendPipe, cancellationToken).ConfigureAwait(false);
        }

        public async Task Send<T>(object values, IPipe<SendContext<T>> pipe, CancellationToken cancellationToken)
            where T : class
        {
            if (values == null)
                throw new ArgumentNullException(nameof(values));
            if (pipe == null)
                throw new ArgumentNullException(nameof(pipe));

            (var message, IPipe<SendContext<T>> sendPipe) =
                await MessageInitializerCache<T>.InitializeMessage(values, GetPipeProxy(pipe), cancellationToken).ConfigureAwait(false);

            await _endpoint.Send(message, sendPipe, cancellationToken).ConfigureAwait(false);
        }

        public async Task Send<T>(object values, IPipe<SendContext> pipe, CancellationToken cancellationToken)
            where T : class
        {
            if (values == null)
                throw new ArgumentNullException(nameof(values));
            if (pipe == null)
                throw new ArgumentNullException(nameof(pipe));

            (var message, IPipe<SendContext<T>> sendPipe) =
                await MessageInitializerCache<T>.InitializeMessage(values, GetPipeProxy<T>(pipe), cancellationToken).ConfigureAwait(false);

            await _endpoint.Send(message, sendPipe, cancellationToken).ConfigureAwait(false);
        }

        protected abstract IPipe<SendContext<T>> GetPipeProxy<T>(IPipe<SendContext<T>>? pipe = default)
            where T : class;
    }
}