Query filter: replace EventEmitter with Observables (#36087) (#36750)

* Changed query filter to use observables instead of event emitter
* Updated tests accordingly

src/legacy/core_plugins/input_control_vis/public/vis_controller.js

@@ -31,7 +31,9 @@ class VisController {
     this.controls = [];
 
     this.queryBarUpdateHandler = this.updateControlsFromKbn.bind(this);
-    this.vis.API.queryFilter.on('update', this.queryBarUpdateHandler);
+
+    this.updateSubsciption = this.vis.API.queryFilter.getUpdates$()
+      .subscribe(this.queryBarUpdateHandler);
   }
 
   async render(visData, visParams, status) {
@@ -46,7 +48,7 @@ class VisController {
   }
 
   destroy() {
-    this.vis.API.queryFilter.off('update', this.queryBarUpdateHandler);
+    this.updateSubsciption.unsubscribe();
     unmountComponentAtNode(this.el);
   }
 

src/legacy/core_plugins/kibana/public/context/index.js

@@ -20,7 +20,6 @@
 import _ from 'lodash';
 
 import { FilterBarQueryFilterProvider } from 'ui/filter_bar/query_filter';
-import 'ui/listen';
 import uiRoutes from 'ui/routes';
 import { i18n } from '@kbn/i18n';
 
@@ -79,8 +78,14 @@ function ContextAppRouteController(
     'contextAppRoute.state.successorCount',
   ], () => this.state.save(true));
 
-  $scope.$listen(queryFilter, 'update', () => {
-    this.filters = _.cloneDeep(queryFilter.getFilters());
+  const updateSubsciption = queryFilter.getUpdates$().subscribe({
+    next: () => {
+      this.filters = _.cloneDeep(queryFilter.getFilters());
+    }
+  });
+
+  $scope.$on('$destroy', function () {
+    updateSubsciption.unsubscribe();
   });
 
   this.anchorType = $routeParams.type;

src/legacy/core_plugins/kibana/public/dashboard/dashboard_app.js

@@ -25,7 +25,6 @@ import chrome from 'ui/chrome';
 import { wrapInI18nContext } from 'ui/i18n';
 import { toastNotifications } from 'ui/notify';
 
-import 'ui/listen';
 import 'ui/apply_filters';
 
 import { panelActionsStore } from './store/panel_actions_store';
@@ -519,15 +518,20 @@ app.directive('dashboardApp', function ($injector) {
       updateViewMode(dashboardStateManager.getViewMode());
 
       // update root source when filters update
-      $scope.$listen(queryFilter, 'update', function () {
-        $scope.model.filters = queryFilter.getFilters();
-        dashboardStateManager.applyFilters($scope.model.query, $scope.model.filters);
+      this.updateSubscription = queryFilter.getUpdates$().subscribe({
+        next: () => {
+          $scope.model.filters = queryFilter.getFilters();
+          dashboardStateManager.applyFilters($scope.model.query, $scope.model.filters);
+        }
       });
 
       // update data when filters fire fetch event
-      $scope.$listen(queryFilter, 'fetch', $scope.refresh);
+
+      this.fetchSubscription = queryFilter.getFetches$().subscribe($scope.refresh);
 
       $scope.$on('$destroy', () => {
+        this.updateSubscription.unsubscribe();
+        this.fetchSubscription.unsubscribe();
         dashboardStateManager.destroy();
       });
 

src/legacy/core_plugins/kibana/public/discover/controllers/discover.js

@@ -210,6 +210,9 @@ function discoverController(
     requests: new RequestAdapter()
   };
 
+  let filterUpdateSubscription;
+  let filterFetchSubscription;
+
   timefilter.disableTimeRangeSelector();
   timefilter.disableAutoRefreshSelector();
 
@@ -226,7 +229,11 @@ function discoverController(
 
   // the saved savedSearch
   const savedSearch = $route.current.locals.savedSearch;
-  $scope.$on('$destroy', savedSearch.destroy);
+  $scope.$on('$destroy', () => {
+    savedSearch.destroy();
+    if (filterFetchSubscription) filterFetchSubscription.unsubscribe();
+    if (filterUpdateSubscription) filterUpdateSubscription.unsubscribe();
+  });
 
   const $appStatus = $scope.appStatus = this.appStatus = {
     dirty: !savedSearch.id
@@ -561,21 +568,23 @@ function discoverController(
         });
 
         // update data source when filters update
-        $scope.$listen(queryFilter, 'update', function () {
-          $scope.filters = queryFilter.getFilters();
-          return $scope.updateDataSource().then(function () {
-            $state.save();
-          });
-        });
+        filterUpdateSubscription = queryFilter.getUpdates$().subscribe(
+          () => {
+            $scope.filters = queryFilter.getFilters();
+            $scope.updateDataSource().then(function () {
+              $state.save();
+            });
+          }
+        );
+
+        // fetch data when filters fire fetch event
+        filterFetchSubscription = queryFilter.getFetches$().subscribe($scope.fetch);
 
         // update data source when hitting forward/back and the query changes
         $scope.$listen($state, 'fetch_with_changes', function (diff) {
           if (diff.indexOf('query') >= 0) $scope.fetch();
         });
 
-        // fetch data when filters fire fetch event
-        $scope.$listen(queryFilter, 'fetch', $scope.fetch);
-
         $scope.$watch('opts.timefield', function (timefield) {
           $scope.enableTimeRangeSelector = !!timefield;
         });

src/legacy/core_plugins/kibana/public/visualize/editor/editor.js

@@ -26,7 +26,6 @@ import 'ui/collapsible_sidebar';
 
 import { capabilities } from 'ui/capabilities';
 import 'ui/apply_filters';
-import 'ui/listen';
 import chrome from 'ui/chrome';
 import React from 'react';
 import angular from 'angular';
@@ -420,10 +419,12 @@ function VisEditor(
     $scope.$listenAndDigestAsync(timefilter, 'refreshIntervalUpdate', updateRefreshInterval);
 
     // update the searchSource when filters update
-    $scope.$listen(queryFilter, 'update', function () {
-      $scope.filters = queryFilter.getFilters();
-      $scope.fetch();
-    });
+    const filterUpdateSubscription = queryFilter.getUpdates$().subscribe(
+      () => {
+        $scope.filters = queryFilter.getFilters();
+        $scope.fetch();
+      },
+    );
 
     // update the searchSource when query updates
     $scope.fetch = function () {
@@ -440,6 +441,7 @@ function VisEditor(
       }
       savedVis.destroy();
       stateMonitor.destroy();
+      filterUpdateSubscription.unsubscribe();
     });
 
     if (!$scope.chrome.getVisible()) {

src/legacy/ui/public/filter_bar/__tests__/_add_filters.js

@@ -140,12 +140,15 @@ describe('add filters', function () {
     });
 
     it('should fire the update and fetch events', async function () {
-      const emitSpy = sinon.spy(queryFilter, 'emit');
+      const updateStub = sinon.stub();
+      const fetchStub = sinon.stub();
 
-      const awaitFetch = new Promise(resolve => {
-        queryFilter.on('fetch', () => {
-          resolve();
-        });
+      queryFilter.getUpdates$().subscribe({
+        next: updateStub,
+      });
+
+      queryFilter.getFetches$().subscribe({
+        next: fetchStub,
       });
 
       // set up the watchers, add new filters, and crank the digest loop
@@ -158,10 +161,8 @@ describe('add filters', function () {
       expect(globalState.save.callCount).to.be(1);
 
       // this time, events should be emitted
-      await awaitFetch;
-      expect(emitSpy.callCount).to.be(2);
-      expect(emitSpy.firstCall.args[0]).to.be('update');
-      expect(emitSpy.secondCall.args[0]).to.be('fetch');
+      expect(fetchStub.called);
+      expect(updateStub.called);
     });
   });
 

src/legacy/ui/public/filter_bar/__tests__/_invert_filters.js

@@ -92,7 +92,16 @@ describe('invert filters', function () {
     });
 
     it('should fire the update and fetch events', function () {
-      const emitSpy = sinon.spy(queryFilter, 'emit');
+      const updateStub = sinon.stub();
+      const fetchStub = sinon.stub();
+
+      queryFilter.getUpdates$().subscribe({
+        next: updateStub,
+      });
+
+      queryFilter.getFetches$().subscribe({
+        next: fetchStub,
+      });
       appState.filters = filters;
 
       // set up the watchers
@@ -101,9 +110,8 @@ describe('invert filters', function () {
       // trigger the digest loop to fire the watchers
       $rootScope.$digest();
 
-      expect(emitSpy.callCount).to.be(2);
-      expect(emitSpy.firstCall.args[0]).to.be('update');
-      expect(emitSpy.secondCall.args[0]).to.be('fetch');
+      expect(fetchStub.called);
+      expect(updateStub.called);
     });
   });
 

src/legacy/ui/public/filter_bar/__tests__/_pin_filters.js

@@ -122,15 +122,25 @@ describe('pin filters', function () {
 
 
     it('should only fire the update event', function () {
-      const emitSpy = sinon.spy(queryFilter, 'emit');
+      const updateStub = sinon.stub();
+      const fetchStub = sinon.stub();
+
+      queryFilter.getUpdates$().subscribe({
+        next: updateStub,
+      });
+
+      queryFilter.getFetches$().subscribe({
+        next: fetchStub,
+      });
+
       const filter = appState.filters[1];
       $rootScope.$digest();
 
       queryFilter.pinFilter(filter);
       $rootScope.$digest();
 
-      expect(emitSpy.callCount).to.be(1);
-      expect(emitSpy.firstCall.args[0]).to.be('update');
+      expect(!fetchStub.called);
+      expect(updateStub.called);
     });
   });
 

src/legacy/ui/public/filter_bar/__tests__/_remove_filters.js

@@ -85,16 +85,26 @@ describe('remove filters', function () {
     });
 
     it('should fire the update and fetch events', function () {
-      const emitSpy = sinon.spy(queryFilter, 'emit');
+      const updateStub = sinon.stub();
+      const fetchStub = sinon.stub();
+
+      queryFilter.getUpdates$().subscribe({
+        next: updateStub,
+      });
+
+      queryFilter.getFetches$().subscribe({
+        next: fetchStub,
+      });
+
       appState.filters = filters;
       $rootScope.$digest();
 
       queryFilter.removeFilter(filters[0]);
       $rootScope.$digest();
 
-      expect(emitSpy.callCount).to.be(2);
-      expect(emitSpy.firstCall.args[0]).to.be('update');
-      expect(emitSpy.secondCall.args[0]).to.be('fetch');
+      // this time, events should be emitted
+      expect(fetchStub.called);
+      expect(updateStub.called);
     });
 
     it('should remove matching filters', function () {

src/legacy/ui/public/filter_bar/__tests__/_toggle_filters.js

@@ -91,16 +91,26 @@ describe('toggle filters', function () {
     });
 
     it('should fire the update and fetch events', function () {
-      const emitSpy = sinon.spy(queryFilter, 'emit');
+      const updateStub = sinon.stub();
+      const fetchStub = sinon.stub();
+
+      queryFilter.getUpdates$().subscribe({
+        next: updateStub,
+      });
+
+      queryFilter.getFetches$().subscribe({
+        next: fetchStub,
+      });
+
       appState.filters = filters;
       $rootScope.$digest();
 
       queryFilter.toggleFilter(filters[1]);
       $rootScope.$digest();
 
-      expect(emitSpy.callCount).to.be(2);
-      expect(emitSpy.firstCall.args[0]).to.be('update');
-      expect(emitSpy.secondCall.args[0]).to.be('fetch');
+      // this time, events should be emitted
+      expect(fetchStub.called);
+      expect(updateStub.called);
     });
 
     it('should always enable the filter', function () {

src/legacy/ui/public/filter_bar/__tests__/query_filter.js

@@ -26,21 +26,19 @@ import './_toggle_filters';
 import './_invert_filters';
 import './_pin_filters';
 import { FilterBarQueryFilterProvider } from '../query_filter';
-import { EventsProvider } from '../../events';
 let queryFilter;
-let EventEmitter;
 
 describe('Query Filter', function () {
   describe('Module', function () {
     beforeEach(ngMock.module('kibana'));
     beforeEach(ngMock.inject(function (_$rootScope_, Private) {
       queryFilter = Private(FilterBarQueryFilterProvider);
-      EventEmitter = Private(EventsProvider);
     }));
 
     describe('module instance', function () {
-      it('should be an event emitter', function () {
-        expect(queryFilter).to.be.an(EventEmitter);
+      it('should use observables', function () {
+        expect(queryFilter.getUpdates$).to.be.a('function');
+        expect(queryFilter.getFetches$).to.be.a('function');
       });
     });
 

src/legacy/ui/public/filter_bar/query_filter.js

@@ -18,19 +18,29 @@
  */
 
 import _ from 'lodash';
+import { Subject } from 'rxjs';
+
 import { onlyDisabled } from './lib/only_disabled';
 import { onlyStateChanged } from './lib/only_state_changed';
 import { uniqFilters } from './lib/uniq_filters';
 import { compareFilters } from './lib/compare_filters';
-import { EventsProvider } from '../events';
 import { mapAndFlattenFilters } from './lib/map_and_flatten_filters';
 import { extractTimeFilter } from './lib/extract_time_filter';
 import { changeTimeFilter } from './lib/change_time_filter';
 
-export function FilterBarQueryFilterProvider(Private, Promise, indexPatterns, $rootScope, getAppState, globalState, config) {
-  const EventEmitter = Private(EventsProvider);
+export function FilterBarQueryFilterProvider(Promise, indexPatterns, $rootScope, getAppState, globalState, config) {
+  const queryFilter = {};
 
-  const queryFilter = new EventEmitter();
+  const update$ = new Subject();
+  const fetch$ = new Subject();
+
+  queryFilter.getUpdates$ = function () {
+    return update$.asObservable();
+  };
+
+  queryFilter.getFetches$ = function () {
+    return fetch$.asObservable();
+  };
 
   queryFilter.getFilters = function () {
     const compareOptions = { disabled: true, negate: true };
@@ -366,15 +376,15 @@ export function FilterBarQueryFilterProvider(Private, Promise, indexPatterns, $r
 
         // check for actions, bail if we're done
         getActions();
-        if (!doUpdate) return;
+        if (doUpdate) {
+          // save states and emit the required events
+          saveState();
+          update$.next();
 
-        // save states and emit the required events
-        saveState();
-        queryFilter.emit('update')
-          .then(function () {
-            if (!doFetch) return;
-            queryFilter.emit('fetch');
-          });
+          if (doFetch) {
+            fetch$.next();
+          }
+        }
 
         // iterate over each state type, checking for changes
         function getActions() {