caSTLe-windsor - 如何将依赖项传递给实现 IEventProcessor(Event Hub) 的类

Question

我有以下问题。

我们使用事件中心。在下面的类中，我们从 IEventProcessor 继承，如您所见，我们使用 Service Locator 。我们不能让它与构造函数/属性注入(inject)一起工作。当旨在从 IEventProcessor 继承的类时，CaSTLe Windsor 似乎无法解决依赖关系。这是一个已知问题还是我需要做些什么才能让它工作？

下面是代码:

public class EventProcessor : IEventProcessor
{
    private readonly IEventService _eventService;
    private readonly ILogger _logger;
    private readonly Lazy<RetryPolicy> _retryPolicy;
    private readonly IConfigurationProvider _configurationProvider;

    public EventProcessor()
    {
        try
        {
            _eventService = ContainerProvider.Current.Container.Resolve<IEventService>();
            _logger = ContainerProvider.Current.Container.Resolve<ILogger>();
            _configurationProvider =     ContainerProvider.Current.Container.Resolve<IConfigurationProvider>();

        }
        catch (Exception exception)
        {
            _logger.WriteError(string.Format("Error occured when intializing EventProcessor: '{0}'", exception));
        }
    }

    public Task OpenAsync(PartitionContext context)
    {
        return Task.FromResult(0);
    }

    public async Task ProcessEventsAsync(PartitionContext context, IEnumerable<EventData> events)
    {
        var eventsList = events.ToList();
        EventData lastEvent = null;
        foreach (var eventData in eventsList)
        {
            _logger.WriteVerbose(string.Format("Consumming {0} events...", eventsList.Count()));
            _eventService.ProcessEvent(eventData);
            lastEvent = eventData;
        }

        if (lastEvent != null)
        {
            await AzureServiceBusRetryPolicy.ExecuteAsync(async () => await context.CheckpointAsync(lastEvent));
        }
    }

    public async Task CloseAsync(PartitionContext context, CloseReason reason)
    {
        _logger.WriteInfo("EventHub processor was closed for this reason: " + reason);

        if (reason == CloseReason.Shutdown)
        {
            await AzureServiceBusRetryPolicy.ExecuteAsync(async () => await context.CheckpointAsync());
        }

    }


}

谢谢

Answer 1

我正在使用 Autofac 但遇到了同样的问题。

我通过实现 IEventProcessorFactory 解决了并在 EventProcessorHost 中注册处理器时使用它.

举个例子，我的EventProcessorHost看起来像这样:

public class EventHubProcessorHost
{
    private readonly IEventProcessorFactory _eventProcessorFactory;
    private readonly string _serviceBusConnectionString;
    private readonly string _storageConnectionString;
    private readonly string _eventHubName;

    public EventHubProcessorHost(IEventProcessorFactory eventProcessorFactory, string serviceBusConnectionString, string storageConnectionString, string eventHubName)
    {
        _eventProcessorFactory = eventProcessorFactory;
        _serviceBusConnectionString = serviceBusConnectionString;
        _storageConnectionString = storageConnectionString;
        _eventHubName = eventHubName;
    }

    public void Start()
    {
        var builder = new ServiceBusConnectionStringBuilder(_serviceBusConnectionString)
        {
            TransportType = TransportType.Amqp
        };

        var client = EventHubClient.CreateFromConnectionString(builder.ToString(), _eventHubName);

        try
        {
            var eventProcessorHost = new EventProcessorHost("singleworker",
              client.Path, client.GetDefaultConsumerGroup().GroupName, builder.ToString(), _storageConnectionString);

            eventProcessorHost.RegisterEventProcessorFactoryAsync(_eventProcessorFactory);
        }
        catch (Exception exp)
        {
            Console.WriteLine("Error on send: " + exp.Message);
        }
    }
}

而且我传入的工厂引用了我的 IoC 容器:

public class MyEventProcessorFactory : IEventProcessorFactory
{
    private readonly IComponentContext _componentContext;

    public MyEventProcessorFactory(IComponentContext componentContext)
    {
        _componentContext = componentContext;
    }

    public IEventProcessor CreateEventProcessor(PartitionContext context)
    {
        return _componentContext.Resolve<IEventProcessor>();
    }
}

这允许我在我的 EventProcessor 中正常使用构造函数注入(inject)。 :

public class MyEventProcessor : IEventProcessor
{
    private IFoo _foo;

    public MyEventProcessor(IFoo foo)
    {
        _foo = foo;
    }

    public Task OpenAsync(PartitionContext context)
    {
        return Task.FromResult<object>(null);
    }

    public async Task ProcessEventsAsync(PartitionContext context, IEnumerable<EventData> events)
    {
        foreach (var eventData in events)
        {
            // Processing code
        }

        await context.CheckpointAsync();
    }

    public async Task CloseAsync(PartitionContext context, CloseReason reason)
    {
        if (reason == CloseReason.Shutdown)
        {
            await context.CheckpointAsync();
        }
    }
}

然后我只是在启动时在 Autofac 容器中正常连接所有内容:

builder.RegisterType<Foo>().As<IFoo>()

builder.RegisterType<MyEventProcessor>().As<IEventProcessor>()

builder.Register(c => new MyEventProcessorFactory(c.Resolve<IComponentContext>())).As<IEventProcessorFactory>();

希望这可以帮助。