fix(layout): breakpoint observer completing unrelated subscribers when preceding subscriber unsubscribes (#14988)

Fixes the `BreakpointObserver` completing subscribers that come after the current one, if one of the preceding subscribers unsubscribes. The issue seems to come from the way we store the listener in order to remove it later on. These changes switch to creating the `Observable` ourselves, rather than going through `fromEventPattern` so that we have more control over the event handler.

Fixes #14983.

Author: crisbeto

src/cdk/layout/BUILD.bazel

@@ -20,6 +20,8 @@ ng_test_library(
   name = "layout_test_sources",
   srcs = glob(["**/*.spec.ts"]),
   deps = [
+    "@rxjs",
+    "@rxjs//operators",
     "//src/cdk/platform",
     ":layout",
   ],

src/cdk/layout/breakpoints-observer.spec.ts

@@ -11,9 +11,11 @@ import {BreakpointObserver, BreakpointState} from './breakpoints-observer';
 import {MediaMatcher} from './media-matcher';
 import {fakeAsync, TestBed, inject, flush} from '@angular/core/testing';
 import {Injectable} from '@angular/core';
+import {Subscription} from 'rxjs';
+import {take} from 'rxjs/operators';
 
 describe('BreakpointObserver', () => {
-  let breakpointManager: BreakpointObserver;
+  let breakpointObserver: BreakpointObserver;
   let mediaMatcher: FakeMediaMatcher;
 
   beforeEach(fakeAsync(() => {
@@ -26,7 +28,7 @@ describe('BreakpointObserver', () => {
   beforeEach(inject(
     [BreakpointObserver, MediaMatcher],
     (bm: BreakpointObserver, mm: FakeMediaMatcher) => {
-      breakpointManager = bm;
+      breakpointObserver = bm;
       mediaMatcher = mm;
     }));
 
@@ -36,48 +38,48 @@ describe('BreakpointObserver', () => {
 
   it('retrieves the whether a query is currently matched', fakeAsync(() => {
     const query = 'everything starts as true in the FakeMediaMatcher';
-    expect(breakpointManager.isMatched(query)).toBeTruthy();
+    expect(breakpointObserver.isMatched(query)).toBeTruthy();
   }));
 
   it('reuses the same MediaQueryList for matching queries', fakeAsync(() => {
     expect(mediaMatcher.queryCount).toBe(0);
-    breakpointManager.observe('query1');
+    breakpointObserver.observe('query1');
     expect(mediaMatcher.queryCount).toBe(1);
-    breakpointManager.observe('query1');
+    breakpointObserver.observe('query1');
     expect(mediaMatcher.queryCount).toBe(1);
-    breakpointManager.observe('query2');
+    breakpointObserver.observe('query2');
     expect(mediaMatcher.queryCount).toBe(2);
-    breakpointManager.observe('query1');
+    breakpointObserver.observe('query1');
     expect(mediaMatcher.queryCount).toBe(2);
   }));
 
   it('splits combined query strings into individual matchMedia listeners', fakeAsync(() => {
     expect(mediaMatcher.queryCount).toBe(0);
-    breakpointManager.observe('query1, query2');
+    breakpointObserver.observe('query1, query2');
     expect(mediaMatcher.queryCount).toBe(2);
-    breakpointManager.observe('query1');
+    breakpointObserver.observe('query1');
     expect(mediaMatcher.queryCount).toBe(2);
-    breakpointManager.observe('query2, query3');
+    breakpointObserver.observe('query2, query3');
     expect(mediaMatcher.queryCount).toBe(3);
   }));
 
   it('accepts an array of queries', fakeAsync(() => {
     const queries = ['1 query', '2 query', 'red query', 'blue query'];
-    breakpointManager.observe(queries);
+    breakpointObserver.observe(queries);
     expect(mediaMatcher.queryCount).toBe(queries.length);
   }));
 
   it('completes all events when the breakpoint manager is destroyed', fakeAsync(() => {
     const firstTest = jasmine.createSpy('test1');
-    breakpointManager.observe('test1').subscribe(undefined, undefined, firstTest);
+    breakpointObserver.observe('test1').subscribe(undefined, undefined, firstTest);
     const secondTest = jasmine.createSpy('test2');
-    breakpointManager.observe('test2').subscribe(undefined, undefined, secondTest);
+    breakpointObserver.observe('test2').subscribe(undefined, undefined, secondTest);
 
     flush();
     expect(firstTest).not.toHaveBeenCalled();
     expect(secondTest).not.toHaveBeenCalled();
 
-    breakpointManager.ngOnDestroy();
+    breakpointObserver.ngOnDestroy();
     flush();
 
     expect(firstTest).toHaveBeenCalled();
@@ -87,7 +89,7 @@ describe('BreakpointObserver', () => {
   it('emits an event on the observable when values change', fakeAsync(() => {
     const query = '(width: 999px)';
     let queryMatchState = false;
-    breakpointManager.observe(query).subscribe((state: BreakpointState) => {
+    breakpointObserver.observe(query).subscribe((state: BreakpointState) => {
       queryMatchState = state.matches;
     });
 
@@ -103,7 +105,7 @@ describe('BreakpointObserver', () => {
       const queryOne = '(width: 999px)';
       const queryTwo = '(width: 700px)';
       let state: BreakpointState = {matches: false, breakpoints: {}};
-      breakpointManager.observe([queryOne, queryTwo]).subscribe((breakpoint: BreakpointState) => {
+      breakpointObserver.observe([queryOne, queryTwo]).subscribe((breakpoint: BreakpointState) => {
         state = breakpoint;
       });
 
@@ -120,38 +122,72 @@ describe('BreakpointObserver', () => {
 
   it('emits a true matches state when the query is matched', fakeAsync(() => {
     const query = '(width: 999px)';
-    breakpointManager.observe(query).subscribe();
+    breakpointObserver.observe(query).subscribe();
     mediaMatcher.setMatchesQuery(query, true);
-    expect(breakpointManager.isMatched(query)).toBeTruthy();
+    expect(breakpointObserver.isMatched(query)).toBeTruthy();
   }));
 
   it('emits a false matches state when the query is not matched', fakeAsync(() => {
     const query = '(width: 999px)';
-    breakpointManager.observe(query).subscribe();
+    breakpointObserver.observe(query).subscribe();
     mediaMatcher.setMatchesQuery(query, false);
-    expect(breakpointManager.isMatched(query)).toBeFalsy();
+    expect(breakpointObserver.isMatched(query)).toBeFalsy();
+  }));
+
+  it('should not complete other subscribers when preceding subscriber completes', fakeAsync(() => {
+    const queryOne = '(width: 700px)';
+    const queryTwo = '(width: 999px)';
+    const breakpoint = breakpointObserver.observe([queryOne, queryTwo]);
+    const subscriptions: Subscription[] = [];
+    let emittedValues: number[] = [];
+
+    subscriptions.push(breakpoint.subscribe(() => emittedValues.push(1)));
+    subscriptions.push(breakpoint.pipe(take(1)).subscribe(() => emittedValues.push(2)));
+    subscriptions.push(breakpoint.subscribe(() => emittedValues.push(3)));
+    subscriptions.push(breakpoint.subscribe(() => emittedValues.push(4)));
+
+    mediaMatcher.setMatchesQuery(queryOne, true);
+    mediaMatcher.setMatchesQuery(queryTwo, false);
+    flush();
+
+    expect(emittedValues).toEqual([1, 2, 3, 4]);
+    emittedValues = [];
+
+    mediaMatcher.setMatchesQuery(queryOne, false);
+    mediaMatcher.setMatchesQuery(queryTwo, true);
+    flush();
+
+    expect(emittedValues).toEqual([1, 3, 4]);
+
+    subscriptions.forEach(subscription => subscription.unsubscribe());
   }));
 });
 
 export class FakeMediaQueryList {
   /** The callback for change events. */
-  addListenerCallback?: (mql: MediaQueryListEvent) => void;
+  private _listeners: ((mql: MediaQueryListEvent) => void)[] = [];
 
   constructor(public matches: boolean, public media: string) {}
 
   /** Toggles the matches state and "emits" a change event. */
   setMatches(matches: boolean) {
     this.matches = matches;
-    this.addListenerCallback!(this as any);
+    this._listeners.forEach(listener => listener(this as any));
   }
 
-  /** Registers the callback method for change events. */
+  /** Registers a callback method for change events. */
   addListener(callback: (mql: MediaQueryListEvent) => void) {
-    this.addListenerCallback = callback;
+    this._listeners.push(callback);
   }
 
-  // Noop removal method for testing.
-  removeListener() { }
+  /** Removes a callback method from the change events. */
+  removeListener(callback: (mql: MediaQueryListEvent) => void) {
+    const index = this._listeners.indexOf(callback);
+
+    if (index > -1) {
+      this._listeners.splice(index, 1);
+    }
+  }
 }
 
 @Injectable()

src/cdk/layout/breakpoints-observer.ts

@@ -8,7 +8,7 @@
 
 import {Injectable, NgZone, OnDestroy} from '@angular/core';
 import {MediaMatcher} from './media-matcher';
-import {asapScheduler, combineLatest, fromEventPattern, Observable, Subject} from 'rxjs';
+import {asapScheduler, combineLatest, Observable, Subject, Observer} from 'rxjs';
 import {debounceTime, map, startWith, takeUntil} from 'rxjs/operators';
 import {coerceArray} from '@angular/cdk/coercion';
 
@@ -99,27 +99,24 @@ export class BreakpointObserver implements OnDestroy {
 
     const mql: MediaQueryList = this.mediaMatcher.matchMedia(query);
 
-    // TODO(jelbourn): change this `any` to `MediaQueryListEvent` once Google has upgraded to
-    // TypeScript 3.1 (the type is unavailable before then).
-    let queryListener: any;
-
     // Create callback for match changes and add it is as a listener.
-    const queryObservable = fromEventPattern<MediaQueryList>(
+    const queryObservable = new Observable<MediaQueryList>((observer: Observer<MediaQueryList>) => {
       // Listener callback methods are wrapped to be placed back in ngZone. Callbacks must be placed
       // back into the zone because matchMedia is only included in Zone.js by loading the
       // webapis-media-query.js file alongside the zone.js file.  Additionally, some browsers do not
       // have MediaQueryList inherit from EventTarget, which causes inconsistencies in how Zone.js
       // patches it.
-      (listener: Function) => {
-        queryListener = (e: any) => this.zone.run(() => listener(e));
-        mql.addListener(queryListener);
-      },
-      () => mql.removeListener(queryListener))
-      .pipe(
-        startWith(mql),
-        map((nextMql: MediaQueryList) => ({query, matches: nextMql.matches})),
-        takeUntil(this._destroySubject)
-      );
+      const handler = (e: any) => this.zone.run(() => observer.next(e));
+      mql.addListener(handler);
+
+      return () => {
+        mql.removeListener(handler);
+      };
+    }).pipe(
+      startWith(mql),
+      map((nextMql: MediaQueryList) => ({query, matches: nextMql.matches})),
+      takeUntil(this._destroySubject)
+    );
 
     // Add the MediaQueryList to the set of queries.
     const output = {observable: queryObservable, mql};