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

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};