fix: fix up notification types (#5478)

* fix: fix up notification types

- Ensures Notification methods only accept appropriate arguments
- Ensures Arbitrary objects of the proper shape my be passed through dematerialized
- Utilizes hot path functions to create lightweight objects for internal notification use
- Deprecates the Notification class
- Adds a deprecation message for materialize to notify users that soon the emitted object type will change to not have all of the same methods as Notification
- Updates a few tests that were relying on the shape of Notification instances to pass, as we are now using POJOs internally in the TestScheduler
- Adds dtslint tests for notifications
- Adds dtslint tests to enforce types on dematerialize

BREAKING CHANGE: Notification.createNext(undefined) will no longer return the exact same reference everytime.

BREAKING CHANGE: Type signatures tightened up around Notification and dematerialize

* chore: address comments

spec-dtslint/Notification-spec.ts

@@ -0,0 +1,26 @@
+import { Notification } from 'rxjs';
+
+describe('Notification', () => {
+  // Basic method tests
+  const nextNotification = Notification.createNext('a'); // $ExpectType Notification<string> & NextNotification<string>
+  nextNotification.do((value: number) => {}); // $ExpectError
+  const r = nextNotification.do((value: string) => {}); // $ExpectType void
+  const r1 = nextNotification.observe({ next: value => { } }); // $ExpectType void
+  const r2 = nextNotification.observe({ next: (value: number) => { } }); // $ExpectError
+  const r3 = nextNotification.toObservable(); // $ExpectType Observable<string>
+  const k1 = nextNotification.kind; // $ExpectType "N"
+
+  const completeNotification = Notification.createComplete(); // $ExpectType Notification<never> & CompleteNotification
+  const r4 = completeNotification.do((value: string) => {}); // $ExpectType void
+  const r5 = completeNotification.observe({ next: value => { } }); // $ExpectType void
+  const r6 = completeNotification.observe({ next: (value: number) => { } }); // $ExpectType void
+  const r7 = completeNotification.toObservable(); // $ExpectType Observable<never>
+  const k2 = completeNotification.kind; // $ExpectType "C"
+
+  const errorNotification = Notification.createError(); // $ExpectType Notification<never> & ErrorNotification
+  const r8 = errorNotification.do((value: string) => {}); // $ExpectType void
+  const r9 = errorNotification.observe({ next: value => { } }); // $ExpectType void
+  const r10 = errorNotification.observe({ next: (value: number) => { } }); // $ExpectType void
+  const r11 = errorNotification.toObservable(); // $ExpectType Observable<never>
+  const k3 = errorNotification.kind; // $ExpectType "E"
+});

spec-dtslint/operators/dematerialize-spec.ts

@@ -1,6 +1,7 @@
-import { of, Notification } from 'rxjs';
+import { of, Notification, Observable, ObservableNotification } from 'rxjs';
 import { dematerialize } from 'rxjs/operators';
 
+
 it('should infer correctly', () => {
   const o = of(Notification.createNext('foo')).pipe(dematerialize()); // $ExpectType Observable<string>
 });
@@ -9,6 +10,54 @@ it('should enforce types', () => {
   const o = of(Notification.createNext('foo')).pipe(dematerialize(() => {})); // $ExpectError
 });
 
+it('should enforce types from POJOS', () => {
+  const source = of({
+    kind: 'N' as const,
+    value: 'test'
+  }, {
+    kind: 'N' as const,
+    value: 123
+  },
+  {
+    kind: 'N' as const,
+    value: [true, false]
+  });
+  const o = source.pipe(dematerialize()); // $ExpectType Observable<string | number | boolean[]>
+
+  // NOTE: The `const` is required, because TS doesn't yet have a way to know for certain the
+  // `kind` properties of these objects won't be mutated at runtime.
+  const source2 = of({
+    kind: 'N' as const,
+    value: 1
+  }, {
+    kind: 'C' as const
+  });
+  const o2 = source2.pipe(dematerialize()); // $ExpectType Observable<number>
+
+  const source3 = of({
+    kind: 'C' as const
+  });
+  const o3 = source3.pipe(dematerialize()); // $ExpectType Observable<never>
+
+  const source4 = of({
+    kind: 'E' as const,
+    error: new Error('bad')
+  });
+  const o4 = source4.pipe(dematerialize()); // $ExpectType Observable<never>
+
+  const source5 = of({
+    kind: 'E' as const
+  });
+  const o5 = source5.pipe(dematerialize()); // $ExpectError
+
+
+  // Next notifications should have a value.
+  const source6 = of({
+    kind: 'N' as const
+  });
+  const o6 = source6.pipe(dematerialize()); // $ExpectError
+});
+
 it('should enforce Notification source', () => {
   const o = of('foo').pipe(dematerialize()); // $ExpectError
 });

spec-dtslint/operators/materialize-spec.ts

@@ -2,7 +2,7 @@ import { of } from 'rxjs';
 import { materialize } from 'rxjs/operators';
 
 it('should infer correctly', () => {
-  const o = of('foo').pipe(materialize()); // $ExpectType Observable<Notification<string>>
+  const o = of('foo').pipe(materialize()); // $ExpectType Observable<(Notification<string> & NextNotification<string>) | (Notification<string> & CompleteNotification) | (Notification<string> & ErrorNotification)>
 });
 
 it('should enforce types', () => {

spec/Notification-spec.ts

@@ -81,13 +81,6 @@ describe('Notification', () => {
       expect(first).not.to.equal(second);
     });
 
-    it('should return static next Notification reference without value', () => {
-      const first = Notification.createNext(undefined);
-      const second = Notification.createNext(undefined);
-
-      expect(first).to.equal(second);
-    });
-
     it('should return static complete Notification reference', () => {
       const first = Notification.createComplete();
       const second = Notification.createComplete();
@@ -160,7 +153,7 @@ describe('Notification', () => {
 
     it('should accept observer for error Notification', () => {
       let observed = false;
-      const n = Notification.createError<string>();
+      const n = Notification.createError();
       const observer = Subscriber.create((x?: string) => {
         throw 'should not be called';
       }, (err: any) => {

spec/operators/dematerialize-spec.ts

@@ -1,8 +1,8 @@
-import { of, Notification } from 'rxjs';
-import { dematerialize, map, mergeMap } from 'rxjs/operators';
+import { of, Notification, ObservableNotification } from 'rxjs';
+import { dematerialize, map, mergeMap, materialize } from 'rxjs/operators';
 import { hot, cold, expectObservable, expectSubscriptions } from '../helpers/marble-testing';
 
-const NO_VALUES: { [key: string]: Notification<any> } = {};
+const NO_VALUES: { [key: string]: ObservableNotification<any> } = {};
 
 /** @test {dematerialize} */
 describe('dematerialize operator', () => {
@@ -48,7 +48,7 @@ describe('dematerialize operator', () => {
   });
 
   it('should dematerialize a sad stream', () => {
-    const values: Record<string, Notification<string | undefined>> = {
+    const values = {
       a: Notification.createNext('w'),
       b: Notification.createNext('x'),
       c: Notification.createNext('y'),
@@ -147,7 +147,7 @@ describe('dematerialize operator', () => {
     expectSubscriptions(e1.subscriptions).toBe(e1subs);
   });
 
-  it('should dematerialize and completes when stream compltes with complete notification', () => {
+  it('should dematerialize and completes when stream completes with complete notification', () => {
     const e1 =   hot('----(a|)', { a: Notification.createComplete() });
     const e1subs =   '^   !';
     const expected = '----|';
@@ -164,4 +164,14 @@ describe('dematerialize operator', () => {
     expectObservable(e1.pipe(dematerialize())).toBe(expected);
     expectSubscriptions(e1.subscriptions).toBe(e1subs);
   });
+
+  it('should work with materialize', () => {
+    const source = hot('----a--b---c---d---e----f--|');
+    const expected = '----a--b---c---d---e----f--|';
+    const result = source.pipe(
+      materialize(),
+      dematerialize()
+    );
+    expectObservable(result).toBe(expected);
+  });
 });

spec/operators/groupBy-spec.ts

@@ -3,6 +3,7 @@ import { groupBy, delay, tap, map, take, mergeMap, materialize, skip } from 'rxj
 import { TestScheduler } from 'rxjs/testing';
 import { ReplaySubject, of, GroupedObservable, Observable, Operator, Observer } from 'rxjs';
 import { hot, cold, expectObservable, expectSubscriptions } from '../helpers/marble-testing';
+import { createNotification } from 'rxjs/internal/Notification';
 
 declare const rxTestScheduler: TestScheduler;
 
@@ -547,18 +548,15 @@ describe('groupBy operator', () => {
       .unsubscribedFrame;
 
     const source = e1.pipe(
-      groupBy((val: string) => val.toLowerCase().trim()),
-      map((group: any) => {
+      groupBy((val) => val.toLowerCase().trim()),
+      map((group) => {
         const arr: any[] = [];
 
         const subscription = group.pipe(
-          materialize(),
-          map((notification: Notification) => {
-            return { frame: rxTestScheduler.frame, notification: notification };
-          })
-        ).subscribe((value: any) => {
-            arr.push(value);
-          });
+          phonyMarbelize()
+        ).subscribe((value) => {
+          arr.push(value);
+        });
 
         if (group.key === 'foo') {
           rxTestScheduler.schedule(() => {
@@ -618,10 +616,7 @@ describe('groupBy operator', () => {
         const arr: any[] = [];
 
         const subscription = group.pipe(
-          materialize(),
-          map((notification: Notification) => {
-            return { frame: rxTestScheduler.frame, notification: notification };
-          })
+          phonyMarbelize()
         ).subscribe((value: any) => {
             arr.push(value);
           });
@@ -893,13 +888,10 @@ describe('groupBy operator', () => {
         const arr: any[] = [];
 
         const subscription = group.pipe(
-          materialize(),
-          map((notification: Notification) => {
-            return { frame: rxTestScheduler.frame, notification: notification };
-          })
+          phonyMarbelize()
         ).subscribe((value: any) => {
-            arr.push(value);
-          });
+          arr.push(value);
+        });
 
         rxTestScheduler.schedule(() => {
           subscription.unsubscribe();
@@ -1124,13 +1116,10 @@ describe('groupBy operator', () => {
         const arr: any[] = [];
 
         const subscription = group.pipe(
-          materialize(),
-          map((notification: Notification) => {
-            return { frame: rxTestScheduler.frame, notification: notification };
-          })
+          phonyMarbelize()
         ).subscribe((value: any) => {
-            arr.push(value);
-          });
+          arr.push(value);
+        });
 
         if (group.key === 'foo' && index === 0) {
           rxTestScheduler.schedule(() => {
@@ -1202,13 +1191,10 @@ describe('groupBy operator', () => {
         const arr: any[] = [];
 
         const subscription = group.pipe(
-          materialize(),
-          map((notification: Notification) => {
-            return { frame: rxTestScheduler.frame, notification: notification };
-          })
+          phonyMarbelize()
         ).subscribe((value: any) => {
-            arr.push(value);
-          });
+          arr.push(value);
+        });
 
         const unsubscriptionFrame = hasUnsubscribed[group.key] ?
           unsubscriptionFrames[group.key + '2'] :
@@ -1272,21 +1258,18 @@ describe('groupBy operator', () => {
          (val: string) => val
        ),
        map((group: any) => {
-         const innerNotifications: any[] = [];
-         const subscriptionFrame = subscriptionFrames[group.key];
-
-         rxTestScheduler.schedule(() => {
-           group.pipe(
-              materialize(),
-              map((notification: Notification) => {
-                return { frame: rxTestScheduler.frame, notification: notification };
-              })
-            ).subscribe((value: any) => {
-               innerNotifications.push(value);
-             });
-         }, subscriptionFrame - rxTestScheduler.frame);
-
-         return innerNotifications;
+        const innerNotifications: any[] = [];
+        const subscriptionFrame = subscriptionFrames[group.key];
+
+        rxTestScheduler.schedule(() => {
+          group.pipe(
+            phonyMarbelize()
+          ).subscribe((value: any) => {
+            innerNotifications.push(value);
+          });
+        }, subscriptionFrame - rxTestScheduler.frame);
+
+        return innerNotifications;
        })
       );
 
@@ -1327,13 +1310,10 @@ describe('groupBy operator', () => {
 
         rxTestScheduler.schedule(() => {
           group.pipe(
-            materialize(),
-            map((notification: Notification) => {
-              return { frame: rxTestScheduler.frame, notification: notification };
-            })
+            phonyMarbelize()
           ).subscribe((value: any) => {
-              arr.push(value);
-            });
+            arr.push(value);
+          });
         }, innerSubscriptionFrame - rxTestScheduler.frame);
 
         return arr;
@@ -1377,13 +1357,10 @@ describe('groupBy operator', () => {
 
         rxTestScheduler.schedule(() => {
           group.pipe(
-            materialize(),
-            map((notification: Notification) => {
-              return { frame: rxTestScheduler.frame, notification: notification };
-            })
+            phonyMarbelize()
           ).subscribe((value: any) => {
-              arr.push(value);
-            });
+            arr.push(value);
+          });
         }, innerSubscriptionFrame - rxTestScheduler.frame);
 
         return arr;
@@ -1429,13 +1406,10 @@ describe('groupBy operator', () => {
 
         rxTestScheduler.schedule(() => {
           group.pipe(
-            materialize(),
-            map((notification: Notification) => {
-              return { frame: rxTestScheduler.frame, notification: notification };
-            })
+            phonyMarbelize()
           ).subscribe((value: any) => {
-              arr.push(value);
-            });
+            arr.push(value);
+          });
         }, innerSubscriptionFrame - rxTestScheduler.frame);
 
         return arr;
@@ -1488,3 +1462,21 @@ describe('groupBy operator', () => {
       });
   });
 });
+
+/**
+ * TODO: A helper operator to deal with legacy tests above that could probably be written a different way
+ */
+function phonyMarbelize<T>() {
+  return (source: Observable<T>) => source.pipe(
+    materialize(),
+    map((notification) => {
+      // Because we're hacking some weird inner-observable marbles here, we need
+      // to make sure this is all the same shape as it would be from the TestScheduler
+      // assertions
+      return {
+        frame: rxTestScheduler.frame,
+        notification: createNotification(notification.kind, notification.value, notification.error)
+      };
+    })
+  );
+}