"use strict";

Object.defineProperty(exports, "__esModule", {
  value: true
});
exports.createAggregator = createAggregator;
var _rxjs = require("rxjs");
/*
 * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
 * or more contributor license agreements. Licensed under the Elastic License
 * 2.0; you may not use this file except in compliance with the Elastic License
 * 2.0.
 */

function createAggregator({
  key,
  taskPollingLifecycle,
  config,
  resetMetrics$,
  taskEventFilter,
  metricsAggregator
}) {
  // Resets the aggregators either when the reset interval has passed or
  // a resetMetrics$ event is received
  (0, _rxjs.merge)((0, _rxjs.interval)(config.metrics_reset_interval).pipe((0, _rxjs.map)(() => true)), resetMetrics$.pipe((0, _rxjs.map)(() => true))).subscribe(() => {
    metricsAggregator.reset();
  });
  const taskEvents$ = taskPollingLifecycle.events.pipe((0, _rxjs.filter)(taskEvent => taskEventFilter(taskEvent)), (0, _rxjs.map)(taskEvent => {
    metricsAggregator.processTaskLifecycleEvent(taskEvent);
    return metricsAggregator.collect();
  }));
  return (0, _rxjs.combineLatest)([taskEvents$.pipe((0, _rxjs.startWith)(metricsAggregator.initialMetric()))]).pipe((0, _rxjs.map)(([value]) => {
    return {
      key,
      value
    };
  }));
}