src/MassTransit/DependencyInjection/Configuration/ScopedExecuteActivityPipeSpecificationObserver.cs - develop - vshmidt/masstransit

src/MassTransit/DependencyInjection/Configuration/ScopedExecuteActivityPipeSpecificationObserver.cs

60 строк2 KB

denys.kozhevnikov

Separate scoped ConsumeContext through components

23 май 2023, 23:32

23 май 2023, 23:32e679624

namespace MassTransit.Configuration
{
    using System;
    using DependencyInjection;
    using Internals;
    using Middleware;


    public class ScopedExecuteActivityPipeSpecificationObserver :
        IActivityConfigurationObserver
    {
        readonly IRegistrationContext _context;
        readonly Type _filterType;
        readonly CompositeFilter<Type> _messageTypeFilter;

        public ScopedExecuteActivityPipeSpecificationObserver(Type filterType, IRegistrationContext context,
            CompositeFilter<Type> messageTypeFilter)
        {
            _filterType = filterType;
            _context = context;
            _messageTypeFilter = messageTypeFilter;
        }

        public void ActivityConfigured<TActivity, TArguments>(IExecuteActivityConfigurator<TActivity, TArguments> configurator, Uri compensateAddress)
            where TActivity : class, IExecuteActivity<TArguments>
            where TArguments : class
        {
            ExecuteActivityConfigured(configurator);
        }

        public void ExecuteActivityConfigured<TActivity, TArguments>(IExecuteActivityConfigurator<TActivity, TArguments> configurator)
            where TActivity : class, IExecuteActivity<TArguments>
            where TArguments : class
        {
            if (!_messageTypeFilter.Matches(typeof(TArguments)))
                return;

            var filterType = _filterType.MakeGenericType(typeof(TArguments));

            if (!filterType.HasInterface(typeof(IFilter<ExecuteContext<TArguments>>)))
                throw new ConfigurationException($"The scoped filter must implement {TypeCache<IFilter<ExecuteContext<TArguments>>>.ShortName} ");

            var scopeProvider = new ExecuteActivityScopeProvider<TActivity, TArguments>(_context);

            var scopedFilterType = typeof(ScopedExecuteFilter<,,>).MakeGenericType(typeof(TActivity), typeof(TArguments), filterType);

            var filter = (IFilter<ExecuteContext<TArguments>>)Activator.CreateInstance(scopedFilterType, scopeProvider);

            var specification = new FilterPipeSpecification<ExecuteContext<TArguments>>(filter);

            configurator.Arguments(x => x.AddPipeSpecification(specification));
        }

        public void CompensateActivityConfigured<TActivity, TLog>(ICompensateActivityConfigurator<TActivity, TLog> configurator)
            where TActivity : class, ICompensateActivity<TLog>
            where TLog : class
        {
        }
    }
}