1

Здравствуйте есть сервис, который поставляет данные с внешнего источника примерно раз в секунду. Поскольку "льется" поток данных то получение данных я обернул в IObservable<MarketingRowData>, для удобства обработки. Эти данные, на начальном этапе, нужно просто сохранять в БД.

Сервис работающий на фоне MarketingRowDataHandlerBg - регистрируется в IoC как SingleInstance. В качестве зависимостей он принимает:

  IMarketingRowDataRepository - InstancePerLifetimeScope
  IUnitOfWork - InstancePerLifetimeScope
  MarketingClientPollingWrapper - SingleInstance

Создаем наблюдателя и подписываемся на поток данных, в обработчике наблюдателя (_marketingCalcObserver) сохраняем полученные данные в БД.

public class MarketingRowDataHandlerBg
{
    private readonly MarketingClientPollingWrapper _marketingClientPolling;
    private readonly MarketingCalcObserver _marketingCalcObserver;
    private IDisposable _marketingCalcLifeTime;


    public MarketingRowDataHandlerBg(
        IMarketingRowDataRepository marketingRep,
        IUnitOfWork uow,
        MarketingClientPollingWrapper marketingClientPolling,
        ILogger logger)
    {
        _marketingClientPolling = marketingClientPolling;
        _marketingCalcObserver = new MarketingCalcObserver(marketingRep, uow, logger);
    }

         
    /// <summary>
    /// Подписка на поток получения данных.
    /// </summary>
    public Task StartAsync(CancellationToken ct)
    {
        _marketingCalcLifeTime= _marketingClientPolling.DataRx
            .TakeWhile(v=>!ct.IsCancellationRequested)
            .Subscribe(_marketingCalcObserver);
        return Task.CompletedTask;
    }
    
    /// <summary>
    /// Отписка от потока получения данных.
    /// </summary>
    public Task StopAsync(CancellationToken ct)
    {
        _marketingCalcLifeTime.Dispose();
        return Task.CompletedTask;
    }
}

internal class MarketingCalcObserver : IObserver<MarketingRowData>
{
    private readonly IMarketingRowDataRepository _marketingRep;
    private readonly IUnitOfWork _uow;
    private readonly ILogger _logger;


    public MarketingCalcObserver(IMarketingRowDataRepository marketingRep, IUnitOfWork uow, ILogger logger)
    {
        _marketingRep = marketingRep;
        _uow = uow;
        _logger = logger;
    }
    
    
    public void OnCompleted()
    {
       
    }

    public void OnError(Exception error)
    {
        _logger.Error(error.Message);
    }

    public void OnNext(MarketingRowData value) //Пока обработчик не асинхронный
    {
        _marketingRep.AddAsync(value).GetAwaiter().GetResult();
        _uow.CommitChangesAsync().GetAwaiter().GetResult();;
        
        _logger.Information(value.Photo.Name);
    }
}

Получается Singleton принимает короткоживущие зависимости, и по идее, надо использовать фабрики для всех InstancePerLifetimeScope, т.е. внедрять Func<IUnitOfWork> uowFactory и также через фабрику репозиторий?

Надо ли периодически закрывать контекст в обработчике?

   public void OnNext(MarketingRowData value) //Пока обработчик не асинхронный
    {
      _marketingRep.AddAsync(value).GetAwaiter().GetResult();
      _uow.CommitChangesAsync().GetAwaiter().GetResult();
      _logger.Information(value.Photo.Name);
    }

Думаю получать данные Батчами по 10 элементов например и сохранять весь батч за один раз, тогда контекст можно создавать через фабрику только раз в 10 сек на момент сохранения. Как лучше поступить? Можно ли на все время жизни программы работать с одним контекстом в фоновом обработчике?

ДОПОЛНЮ модулем регистрации всех зависимостей EFCore через Autofac

 public class DataAccessAutofacModule : Module
    {
        private readonly string _connectionString;

        public DataAccessAutofacModule(string connectionString)
        {
            _connectionString = connectionString;
            if (string.IsNullOrEmpty(connectionString)) throw new ArgumentException("Connection string to database must be set.");
        }


        protected override void Load(ContainerBuilder builder)
        {
            RegisterContext(builder);
            RegisterUoW(builder);
            RegisterHelpersDb(builder);
            RegisterRepositories(builder);
        }
        
        private void RegisterContext(ContainerBuilder builder)
        {
            builder.Register(c =>
                {
                    var contextBuilder = new DbContextOptionsBuilder<LoanDbContext>();
                    contextBuilder.UseNpgsql(_connectionString);
                    return new LoanDbContext(contextBuilder.Options);
                })
                .AsSelf()
                .InstancePerLifetimeScope();
        }
        
        
        private static void RegisterUoW(ContainerBuilder builder)
        {
            builder
                .RegisterType<EfUnitOfWork>().As<IUnitOfWork>()
                .InstancePerLifetimeScope();
        }
        
        
        private static void RegisterHelpersDb(ContainerBuilder builder)
        {
            builder
                .RegisterType<EfHelpersDb>().As<IHelpersDb>()
                .InstancePerLifetimeScope();
        }
        
        private static void RegisterRepositories(ContainerBuilder builder)
        {
            builder
                .RegisterType<EfOperatorRepository>().As<IOperatorRepository>()
                .InstancePerLifetimeScope();

            builder
                .RegisterType<EfLoanApplicationRepository>().As<ILoanApplicationRepository>()
                .InstancePerLifetimeScope();
            
            builder
                .RegisterType<EfMarketingRowDataRepository>().As<IMarketingRowDataRepository>()
                .InstancePerLifetimeScope();
        }
    }
7
  • 1
    Не совсем понятна ваша реализация. Почему не используете для контекста services.AddDbContextPool<T>() а для репозитория services.AddScoped<T>(). Да и зачем нужны фабрики, если вы используете IoC контейнер?
    – Bulson
    Commented 25 нояб. 2020 в 7:40
  • MarketingRowDataHandlerBg уже будет Singleton он регистрируется через IHostedService. Я использую Autofac в качестве IoC, поэтому регистрация через AddDbContextPool<T>() не доступна.
    – Aldmi
    Commented 25 нояб. 2020 в 7:52
  • К вопросу, добавил код регистрации контекста.
    – Aldmi
    Commented 25 нояб. 2020 в 7:57
  • Bulson - Если бы контекст попадал кол мне из пула, то вопрос был бы такой-же, я бы на все время его захватил и обратно в пул не отдал
    – Aldmi
    Commented 25 нояб. 2020 в 7:59
  • 2
    Связанный вопрос: ru.stackoverflow.com/q/790154/213987
    – A K
    Commented 25 нояб. 2020 в 8:09

0

Ваш ответ

Нажимая «Отправить ответ», вы соглашаетесь с условиями пользования и подтверждаете, что прочитали политику конфиденциальности.

Посмотрите другие вопросы с метками или задайте свой вопрос.