diff --git a/spec-dtslint/Notification-spec.ts b/spec-dtslint/Notification-spec.ts
new file mode 100644
index 0000000000..4d915f7bd6
--- /dev/null
+++ b/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"
+});
\ No newline at end of file
diff --git a/spec-dtslint/operators/dematerialize-spec.ts b/spec-dtslint/operators/dematerialize-spec.ts
index 90eebdc137..b4850be3eb 100644
--- a/spec-dtslint/operators/dematerialize-spec.ts
+++ b/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
 });
diff --git a/spec-dtslint/operators/materialize-spec.ts b/spec-dtslint/operators/materialize-spec.ts
index 5d1f16c1b9..b2a6a341cc 100644
--- a/spec-dtslint/operators/materialize-spec.ts
+++ b/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', () => {
diff --git a/spec/Notification-spec.ts b/spec/Notification-spec.ts
index 34778c9327..7b484880c0 100644
--- a/spec/Notification-spec.ts
+++ b/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) => {
diff --git a/spec/operators/dematerialize-spec.ts b/spec/operators/dematerialize-spec.ts
index d4a4aecd5c..54b26c09bd 100644
--- a/spec/operators/dematerialize-spec.ts
+++ b/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);
+  });
 });
diff --git a/spec/operators/groupBy-spec.ts b/spec/operators/groupBy-spec.ts
index 347d5d78d1..ac27dbec93 100644
--- a/spec/operators/groupBy-spec.ts
+++ b/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)
+      };
+    })
+  );
+}
\ No newline at end of file
diff --git a/spec/schedulers/TestScheduler-spec.ts b/spec/schedulers/TestScheduler-spec.ts
index d0cf8e920b..9ae453dac9 100644
--- a/spec/schedulers/TestScheduler-spec.ts
+++ b/spec/schedulers/TestScheduler-spec.ts
@@ -2,8 +2,9 @@ import { expect } from 'chai';
 import { hot, cold, expectObservable, expectSubscriptions, time } from '../helpers/marble-testing';
 import { AsyncScheduler } from 'rxjs/internal/scheduler/AsyncScheduler';
 import { TestScheduler } from 'rxjs/testing';
-import { Observable, NEVER, EMPTY, Subject, of, merge, Notification } from 'rxjs';
+import { Observable, NEVER, EMPTY, Subject, of, merge } from 'rxjs';
 import { delay, debounceTime, concatMap } from 'rxjs/operators';
+import { nextNotification, COMPLETE_NOTIFICATION, errorNotification } from 'rxjs/internal/Notification';
 
 declare const rxTestScheduler: TestScheduler;
 
@@ -22,54 +23,54 @@ describe('TestScheduler', () => {
     it('should parse a marble string into a series of notifications and types', () => {
       const result = TestScheduler.parseMarbles('-------a---b---|', { a: 'A', b: 'B' });
       expect(result).deep.equal([
-        { frame: 70, notification: Notification.createNext('A') },
-        { frame: 110, notification: Notification.createNext('B') },
-        { frame: 150, notification: Notification.createComplete() }
+        { frame: 70, notification: nextNotification('A') },
+        { frame: 110, notification: nextNotification('B') },
+        { frame: 150, notification: COMPLETE_NOTIFICATION }
       ]);
     });
 
     it('should parse a marble string, allowing spaces too', () => {
       const result = TestScheduler.parseMarbles('--a--b--|   ', { a: 'A', b: 'B' });
       expect(result).deep.equal([
-        { frame: 20, notification: Notification.createNext('A') },
-        { frame: 50, notification: Notification.createNext('B') },
-        { frame: 80, notification: Notification.createComplete() }
+        { frame: 20, notification: nextNotification('A') },
+        { frame: 50, notification: nextNotification('B') },
+        { frame: 80, notification: COMPLETE_NOTIFICATION }
       ]);
     });
 
     it('should parse a marble string with a subscription point', () => {
       const result = TestScheduler.parseMarbles('---^---a---b---|', { a: 'A', b: 'B' });
       expect(result).deep.equal([
-        { frame: 40, notification: Notification.createNext('A') },
-        { frame: 80, notification: Notification.createNext('B') },
-        { frame: 120, notification: Notification.createComplete() }
+        { frame: 40, notification: nextNotification('A') },
+        { frame: 80, notification: nextNotification('B') },
+        { frame: 120, notification: COMPLETE_NOTIFICATION }
       ]);
     });
 
     it('should parse a marble string with an error', () => {
       const result = TestScheduler.parseMarbles('-------a---b---#', { a: 'A', b: 'B' }, 'omg error!');
       expect(result).deep.equal([
-        { frame: 70, notification: Notification.createNext('A') },
-        { frame: 110, notification: Notification.createNext('B') },
-        { frame: 150, notification: Notification.createError('omg error!') }
+        { frame: 70, notification: nextNotification('A') },
+        { frame: 110, notification: nextNotification('B') },
+        { frame: 150, notification: errorNotification('omg error!') }
       ]);
     });
 
     it('should default in the letter for the value if no value hash was passed', () => {
       const result = TestScheduler.parseMarbles('--a--b--c--');
       expect(result).deep.equal([
-        { frame: 20, notification: Notification.createNext('a') },
-        { frame: 50, notification: Notification.createNext('b') },
-        { frame: 80, notification: Notification.createNext('c') },
+        { frame: 20, notification: nextNotification('a') },
+        { frame: 50, notification: nextNotification('b') },
+        { frame: 80, notification: nextNotification('c') },
       ]);
     });
 
     it('should handle grouped values', () => {
       const result = TestScheduler.parseMarbles('---(abc)---');
       expect(result).deep.equal([
-        { frame: 30, notification: Notification.createNext('a') },
-        { frame: 30, notification: Notification.createNext('b') },
-        { frame: 30, notification: Notification.createNext('c') }
+        { frame: 30, notification: nextNotification('a') },
+        { frame: 30, notification: nextNotification('b') },
+        { frame: 30, notification: nextNotification('c') }
       ]);
     });
 
@@ -77,10 +78,10 @@ describe('TestScheduler', () => {
       const runMode = true;
       const result = TestScheduler.parseMarbles('  -a - b -    c |       ', { a: 'A', b: 'B', c: 'C' }, undefined, undefined, runMode);
       expect(result).deep.equal([
-        { frame: 10, notification: Notification.createNext('A') },
-        { frame: 30, notification: Notification.createNext('B') },
-        { frame: 50, notification: Notification.createNext('C') },
-        { frame: 60, notification: Notification.createComplete() }
+        { frame: 10, notification: nextNotification('A') },
+        { frame: 30, notification: nextNotification('B') },
+        { frame: 50, notification: nextNotification('C') },
+        { frame: 60, notification: COMPLETE_NOTIFICATION }
       ]);
     });
 
@@ -88,10 +89,10 @@ describe('TestScheduler', () => {
       const runMode = true;
       const result = TestScheduler.parseMarbles('10.2ms a 1.2s b 1m c|', { a: 'A', b: 'B', c: 'C' }, undefined, undefined, runMode);
       expect(result).deep.equal([
-        { frame: 10.2, notification: Notification.createNext('A') },
-        { frame: 10.2 + 10 + (1.2 * 1000), notification: Notification.createNext('B') },
-        { frame: 10.2 + 10 + (1.2 * 1000) + 10 + (1000 * 60), notification: Notification.createNext('C') },
-        { frame: 10.2 + 10 + (1.2 * 1000) + 10 + (1000 * 60) + 10, notification: Notification.createComplete() }
+        { frame: 10.2, notification: nextNotification('A') },
+        { frame: 10.2 + 10 + (1.2 * 1000), notification: nextNotification('B') },
+        { frame: 10.2 + 10 + (1.2 * 1000) + 10 + (1000 * 60), notification: nextNotification('C') },
+        { frame: 10.2 + 10 + (1.2 * 1000) + 10 + (1000 * 60) + 10, notification: COMPLETE_NOTIFICATION }
       ]);
     });
   });
diff --git a/src/internal/Notification.ts b/src/internal/Notification.ts
index dff88c2a0e..1d600f1446 100644
--- a/src/internal/Notification.ts
+++ b/src/internal/Notification.ts
@@ -1,4 +1,10 @@
-import { PartialObserver } from './types';
+import {
+  PartialObserver,
+  ObservableNotification,
+  CompleteNotification,
+  NextNotification,
+  ErrorNotification,
+} from './types';
 import { Observable } from './Observable';
 import { EMPTY } from './observable/empty';
 import { of } from './observable/of';
@@ -25,76 +31,154 @@ export enum NotificationKind {
  * @see {@link materialize}
  * @see {@link dematerialize}
  * @see {@link observeOn}
- *
- * @class Notification<T>
+ * @deprecated remove in v8. It is NOT recommended to create instances of `Notification` directly
+ * and use them. Rather, try to create POJOs matching the signature outlined in {@link ObservableNotification}.
+ * For example: `{ kind: 'N', value: 1 }`, `{kind: 'E', error: new Error('bad') }`, or `{ kind: 'C' }`.
  */
 export class Notification<T> {
-  hasValue: boolean;
+  /**
+   * A value signifying that the notification will "next" if observed. In truth,
+   * This is really synonomous with just checking `kind === "N"`.
+   * @deprecated remove in v8. Instead, just check to see if the value of `kind` is `"N"`.
+   */
+  readonly hasValue: boolean;
 
+  /**
+   * Creates a "Next" notification object.
+   * @param kind Always `'N'`
+   * @param value The value to notify with if observed.
+   * @deprecated internal as of v8. Use {@link createNext} instead.
+   */
   constructor(kind: 'N', value?: T);
+  /**
+   * Creates an "Error" notification object.
+   * @param kind Always `'E'`
+   * @param value Always `undefined`
+   * @param error The error to notify with if observed.
+   * @deprecated internal as of v8. Use {@link createError} instead.
+   */
   constructor(kind: 'E', value: undefined, error: any);
+  /**
+   * Creates a "completion" notification object.
+   * @param kind Always `'C'`
+   * @deprecated internal as of v8. Use {@link createComplete} instead.
+   */
   constructor(kind: 'C');
-  constructor(public kind: 'N' | 'E' | 'C', public value?: T, public error?: any) {
+  constructor(public readonly kind: 'N' | 'E' | 'C', public readonly value?: T, public readonly error?: any) {
     this.hasValue = kind === 'N';
   }
 
   /**
-   * Delivers to the given `observer` the value wrapped by this Notification.
-   * @param {Observer} observer
-   * @return
+   * Executes the appropriate handler on a passed `observer` given the `kind` of notification.
+   * If the handler is missing it will do nothing. Even if the notification is an error, if
+   * there is no error handler on the observer, an error will not be thrown, it will noop.
+   * @param observer The observer to notify.
    */
-  observe(observer: PartialObserver<T>): any {
+  observe(observer: PartialObserver<T>): void {
     switch (this.kind) {
       case 'N':
-        return observer.next && observer.next(this.value!);
+        observer.next?.(this.value!);
+        break;
       case 'E':
-        return observer.error && observer.error(this.error);
+        observer.error?.(this.error);
+        break;
       case 'C':
-        return observer.complete && observer.complete();
+        observer.complete?.();
+        break;
     }
   }
 
   /**
-   * Given some {@link Observer} callbacks, deliver the value represented by the
-   * current Notification to the correctly corresponding callback.
-   * @param {function(value: T): void} next An Observer `next` callback.
-   * @param {function(err: any): void} [error] An Observer `error` callback.
-   * @param {function(): void} [complete] An Observer `complete` callback.
-   * @return {any}
+   * Executes a notification on the appropriate handler from a list provided.
+   * If a handler is missing for the kind of notification, nothing is called
+   * and no error is thrown, it will be a noop.
+   * @param next A next handler
+   * @param error An error handler
+   * @param complete A complete handler
+   * @deprecated remove in v8. use {@link Notification.prototype.observe} instead.
+   */
+  do(next: (value: T) => void, error: (err: any) => void, complete: () => void): void;
+  /**
+   * Executes a notification on the appropriate handler from a list provided.
+   * If a handler is missing for the kind of notification, nothing is called
+   * and no error is thrown, it will be a noop.
+   * @param next A next handler
+   * @param error An error handler
+   * @deprecated remove in v8. use {@link Notification.prototype.observe} instead.
    */
-  do(next: (value: T) => void, error?: (err: any) => void, complete?: () => void): any {
+  do(next: (value: T) => void, error: (err: any) => void): void;
+  /**
+   * Executes the next handler if the Notification is of `kind` `"N"`. Otherwise
+   * this will not error, and it will be a noop.
+   * @param next The next handler
+   * @deprecated remove in v8. use {@link Notification.prototype.observe} instead.
+   */
+  do(next: (value: T) => void): void;
+  do(next: (value: T) => void, error?: (err: any) => void, complete?: () => void): void {
     const kind = this.kind;
     switch (kind) {
       case 'N':
-        return next && next(this.value!);
+        next?.(this.value!);
+        break;
       case 'E':
-        return error && error(this.error);
+        error?.(this.error);
+        break;
       case 'C':
-        return complete && complete();
+        complete?.();
+        break;
     }
   }
 
   /**
-   * Takes an Observer or its individual callback functions, and calls `observe`
-   * or `do` methods accordingly.
-   * @param {Observer|function(value: T): void} nextOrObserver An Observer or
-   * the `next` callback.
-   * @param {function(err: any): void} [error] An Observer `error` callback.
-   * @param {function(): void} [complete] An Observer `complete` callback.
-   * @return {any}
+   * Executes a notification on the appropriate handler from a list provided.
+   * If a handler is missing for the kind of notification, nothing is called
+   * and no error is thrown, it will be a noop.
+   * @param next A next handler
+   * @param error An error handler
+   * @param complete A complete handler
+   * @deprecated remove in v8. use {@link Notification.prototype.observe} instead.
+   */
+  accept(next: (value: T) => void, error: (err: any) => void, complete: () => void): void;
+  /**
+   * Executes a notification on the appropriate handler from a list provided.
+   * If a handler is missing for the kind of notification, nothing is called
+   * and no error is thrown, it will be a noop.
+   * @param next A next handler
+   * @param error An error handler
+   * @deprecated remove in v8. use {@link Notification.prototype.observe} instead.
+   */
+  accept(next: (value: T) => void, error: (err: any) => void): void;
+  /**
+   * Executes the next handler if the Notification is of `kind` `"N"`. Otherwise
+   * this will not error, and it will be a noop.
+   * @param next The next handler
+   * @deprecated remove in v8. use {@link Notification.prototype.observe} instead.
    */
+  accept(next: (value: T) => void): void;
+
+  /**
+   * Executes the appropriate handler on a passed `observer` given the `kind` of notification.
+   * If the handler is missing it will do nothing. Even if the notification is an error, if
+   * there is no error handler on the observer, an error will not be thrown, it will noop.
+   * @param observer The observer to notify.
+   * @deprecated remove in v8. Use {@link Notification.prototype.observe} instead.
+   */
+  accept(observer: PartialObserver<T>): void;
   accept(nextOrObserver: PartialObserver<T> | ((value: T) => void), error?: (err: any) => void, complete?: () => void) {
     if (nextOrObserver && typeof (<PartialObserver<T>>nextOrObserver).next === 'function') {
       return this.observe(<PartialObserver<T>>nextOrObserver);
     } else {
-      return this.do(<(value: T) => void>nextOrObserver, error, complete);
+      return this.do(<(value: T) => void>nextOrObserver, error as any, complete as any);
     }
   }
 
   /**
    * Returns a simple Observable that just delivers the notification represented
    * by this Notification instance.
-   * @return {any}
+   *
+   * @deprecated remove in v8. In order to accomplish converting `Notification` to an {@link Observable}
+   * you may use {@link of} and {@link dematerialize}: `of(notification).pipe(dematerialize())`. This is
+   * being removed as it has limited usefulness, and we're trying to streamline the library.
    */
   toObservable(): Observable<T> {
     const kind = this.kind;
@@ -109,9 +193,7 @@ export class Notification<T> {
     throw new Error('unexpected notification kind value');
   }
 
-  private static completeNotification: Notification<any> = new Notification('C');
-  private static undefinedValueNotification: Notification<any> = new Notification('N', undefined);
-
+  private static completeNotification = new Notification('C') as Notification<never> & CompleteNotification;
   /**
    * A shortcut to create a Notification instance of the type `next` from a
    * given value.
@@ -119,12 +201,12 @@ export class Notification<T> {
    * @return {Notification<T>} The "next" Notification representing the
    * argument.
    * @nocollapse
+   * @deprecated remove in v8. It is NOT recommended to create instances of `Notification` directly
+   * and use them. Rather, try to create POJOs matching the signature outlined in {@link ObservableNotification}.
+   * For example: `{ kind: 'N', value: 1 }`, `{kind: 'E', error: new Error('bad') }`, or `{ kind: 'C' }`.
    */
-  static createNext<T>(value: T): Notification<T> {
-    if (typeof value !== 'undefined') {
-      return new Notification('N', value);
-    }
-    return Notification.undefinedValueNotification;
+  static createNext<T>(value: T) {
+    return new Notification('N', value) as Notification<T> & NextNotification<T>;
   }
 
   /**
@@ -134,17 +216,86 @@ export class Notification<T> {
    * @return {Notification<T>} The "error" Notification representing the
    * argument.
    * @nocollapse
+   * @deprecated remove in v8. It is NOT recommended to create instances of `Notification` directly
+   * and use them. Rather, try to create POJOs matching the signature outlined in {@link ObservableNotification}.
+   * For example: `{ kind: 'N', value: 1 }`, `{kind: 'E', error: new Error('bad') }`, or `{ kind: 'C' }`.
    */
-  static createError<T>(err?: any): Notification<T> {
-    return new Notification('E', undefined, err);
+  static createError(err?: any) {
+    return new Notification('E', undefined, err) as Notification<never> & ErrorNotification;
   }
 
   /**
    * A shortcut to create a Notification instance of the type `complete`.
    * @return {Notification<any>} The valueless "complete" Notification.
    * @nocollapse
+   * @deprecated remove in v8. It is NOT recommended to create instances of `Notification` directly
+   * and use them. Rather, try to create POJOs matching the signature outlined in {@link ObservableNotification}.
+   * For example: `{ kind: 'N', value: 1 }`, `{kind: 'E', error: new Error('bad') }`, or `{ kind: 'C' }`.
    */
-  static createComplete(): Notification<any> {
+  static createComplete(): Notification<never> & CompleteNotification {
     return Notification.completeNotification;
   }
 }
+
+/**
+ * Executes the appropriate handler on a passed `observer` given the `kind` of notification.
+ * If the handler is missing it will do nothing. Even if the notification is an error, if
+ * there is no error handler on the observer, an error will not be thrown, it will noop.
+ * @param notification The notification object to observe.
+ * @param observer The observer to notify.
+ */
+export function observeNotification<T>(notification: ObservableNotification<T>, observer: PartialObserver<T>) {
+  if (typeof notification.kind !== 'string') {
+    throw new TypeError('Invalid notification, missing "kind"');
+  }
+  switch (notification.kind) {
+    case 'N':
+      observer.next?.(notification.value!);
+      break;
+    case 'E':
+      observer.error?.(notification.error);
+      break;
+    case 'C':
+      observer.complete?.();
+      break;
+  }
+}
+
+/**
+ * A completion object optimized for memory use and created to be the
+ * same "shape" as other notifications in v8.
+ * @internal
+ */
+export const COMPLETE_NOTIFICATION = (() => createNotification('C', undefined, undefined) as CompleteNotification)();
+
+/**
+ * Internal use only. Creates an optimized error notification that is the same "shape"
+ * as other notifications.
+ * @internal
+ */
+export function errorNotification(error: any): ErrorNotification {
+  return createNotification('E', undefined, error) as any;
+}
+
+/**
+ * Internal use only. Creates an optimized next notification that is the same "shape"
+ * as other notifications.
+ * @internal
+ */
+export function nextNotification<T>(value: T) {
+  return createNotification('N', value, undefined) as NextNotification<T>;
+}
+
+/**
+ * Ensures that all notifications created internally have the same "shape" in v8.
+ *
+ * TODO: This is only exported to support a crazy legacy test in `groupBy`.
+ * @internal
+ */
+export function createNotification(kind: 'N' | 'E' | 'C', value: any, error: any) {
+  return {
+    kind,
+    value,
+    error,
+  };
+}
diff --git a/src/internal/operators/delay.ts b/src/internal/operators/delay.ts
index 1156778953..7a232507a7 100644
--- a/src/internal/operators/delay.ts
+++ b/src/internal/operators/delay.ts
@@ -2,10 +2,13 @@ import { async } from '../scheduler/async';
 import { isValidDate } from '../util/isDate';
 import { Operator } from '../Operator';
 import { Subscriber } from '../Subscriber';
-import { Subscription } from '../Subscription';
-import { Notification } from '../Notification';
 import { Observable } from '../Observable';
-import { MonoTypeOperatorFunction, PartialObserver, SchedulerAction, SchedulerLike, TeardownLogic } from '../types';
+import {
+  MonoTypeOperatorFunction,
+  SchedulerAction,
+  SchedulerLike,
+  TeardownLogic
+} from '../types';
 
 /**
  * Delays the emission of items from the source Observable by a given timeout or
@@ -54,19 +57,14 @@ import { MonoTypeOperatorFunction, PartialObserver, SchedulerAction, SchedulerLi
  * managing the timers that handle the time-shift for each item.
  * @return {Observable} An Observable that delays the emissions of the source
  * Observable by the specified timeout or Date.
- * @name delay
  */
-export function delay<T>(delay: number|Date,
-                         scheduler: SchedulerLike = async): MonoTypeOperatorFunction<T> {
-  const absoluteDelay = isValidDate(delay);
-  const delayFor = absoluteDelay ? (+delay - scheduler.now()) : Math.abs(<number>delay);
+export function delay<T>(delay: number | Date, scheduler: SchedulerLike = async): MonoTypeOperatorFunction<T> {
+  const delayFor = isValidDate(delay) ? +delay - scheduler.now() : Math.abs(delay);
   return (source: Observable<T>) => source.lift(new DelayOperator(delayFor, scheduler));
 }
 
 class DelayOperator<T> implements Operator<T, T> {
-  constructor(private delay: number,
-              private scheduler: SchedulerLike) {
-  }
+  constructor(private delay: number, private scheduler: SchedulerLike) {}
 
   call(subscriber: Subscriber<T>, source: any): TeardownLogic {
     return source.subscribe(new DelaySubscriber(subscriber, this.delay, this.scheduler));
@@ -75,7 +73,7 @@ class DelayOperator<T> implements Operator<T, T> {
 
 interface DelayState<T> {
   source: DelaySubscriber<T>;
-  destination: PartialObserver<T>;
+  destination: Subscriber<T>;
   scheduler: SchedulerLike;
 }
 
@@ -87,7 +85,6 @@ interface DelayState<T> {
 class DelaySubscriber<T> extends Subscriber<T> {
   private queue: Array<DelayMessage<T>> = [];
   private active: boolean = false;
-  private errored: boolean = false;
 
   private static dispatch<T>(this: SchedulerAction<DelayState<T>>, state: DelayState<T>): void {
     const source = state.source;
@@ -95,8 +92,8 @@ class DelaySubscriber<T> extends Subscriber<T> {
     const scheduler = state.scheduler;
     const destination = state.destination;
 
-    while (queue.length > 0 && (queue[0].time - scheduler.now()) <= 0) {
-      queue.shift()!.notification.observe(destination);
+    while (queue.length > 0 && queue[0].time - scheduler.now() <= 0) {
+      destination.next(queue.shift()!.value);
     }
 
     if (queue.length > 0) {
@@ -111,41 +108,34 @@ class DelaySubscriber<T> extends Subscriber<T> {
     }
   }
 
-  constructor(destination: Subscriber<T>,
-              private delay: number,
-              private scheduler: SchedulerLike) {
+  constructor(protected destination: Subscriber<T>, private delay: number, private scheduler: SchedulerLike) {
     super(destination);
   }
 
   private _schedule(scheduler: SchedulerLike): void {
     this.active = true;
-    const destination = this.destination as Subscription;
-    destination.add(scheduler.schedule<DelayState<T>>(DelaySubscriber.dispatch as any, this.delay, {
-      source: this, destination: this.destination, scheduler: scheduler
-    }));
+    const { destination } = this;
+    // TODO: The cast below seems like an issue with typings for SchedulerLike to me.
+    destination.add(
+      scheduler.schedule<DelayState<T>>(DelaySubscriber.dispatch as any, this.delay, {
+        source: this,
+        destination,
+        scheduler,
+      } as DelayState<T>)
+    );
   }
 
-  private scheduleNotification(notification: Notification<T>): void {
-    if (this.errored === true) {
-      return;
-    }
-
+  protected _next(value: T) {
     const scheduler = this.scheduler;
-    const message = new DelayMessage(scheduler.now() + this.delay, notification);
+    const message = new DelayMessage(scheduler.now() + this.delay, value);
     this.queue.push(message);
-
     if (this.active === false) {
       this._schedule(scheduler);
     }
   }
 
-  protected _next(value: T) {
-    this.scheduleNotification(Notification.createNext(value));
-  }
-
   protected _error(err: any) {
-    this.errored = true;
-    this.queue = [];
+    this.queue.length = 0;
     this.destination.error(err);
     this.unsubscribe();
   }
@@ -159,7 +149,5 @@ class DelaySubscriber<T> extends Subscriber<T> {
 }
 
 class DelayMessage<T> {
-  constructor(public readonly time: number,
-              public readonly notification: Notification<T>) {
-  }
+  constructor(public readonly time: number, public readonly value: T) {}
 }
diff --git a/src/internal/operators/dematerialize.ts b/src/internal/operators/dematerialize.ts
index 254dbaa657..4e8252db82 100644
--- a/src/internal/operators/dematerialize.ts
+++ b/src/internal/operators/dematerialize.ts
@@ -1,20 +1,20 @@
 import { Operator } from '../Operator';
 import { Observable } from '../Observable';
 import { Subscriber } from '../Subscriber';
-import { Notification } from '../Notification';
-import { OperatorFunction } from '../types';
+import { observeNotification } from '../Notification';
+import { OperatorFunction, ObservableNotification, ValueFromNotification } from '../types';
 
 /**
- * Converts an Observable of {@link Notification} objects into the emissions
+ * Converts an Observable of {@link ObservableNotification} objects into the emissions
  * that they represent.
  *
- * <span class="informal">Unwraps {@link Notification} objects as actual `next`,
+ * <span class="informal">Unwraps {@link ObservableNotification} objects as actual `next`,
  * `error` and `complete` emissions. The opposite of {@link materialize}.</span>
  *
  * ![](dematerialize.png)
  *
  * `dematerialize` is assumed to operate an Observable that only emits
- * {@link Notification} objects as `next` emissions, and does not emit any
+ * {@link ObservableNotification} objects as `next` emissions, and does not emit any
  * `error`. Such Observable is the output of a `materialize` operation. Those
  * notifications are then unwrapped using the metadata they contain, and emitted
  * as `next`, `error`, and `complete` on the output Observable.
@@ -22,42 +22,44 @@ import { OperatorFunction } from '../types';
  * Use this operator in conjunction with {@link materialize}.
  *
  * ## Example
+ *
  * Convert an Observable of Notifications to an actual Observable
+ *
  * ```ts
- * import { of, Notification } from 'rxjs';
+ * import { of } from 'rxjs';
  * import { dematerialize } from 'rxjs/operators';
  *
- * const notifA = new Notification('N', 'A');
- * const notifB = new Notification('N', 'B');
- * const notifE = new Notification('E', undefined,
- *   new TypeError('x.toUpperCase is not a function')
- * );
+ * const notifA = { kind: 'N', value: 'A' };
+ * const notifB = { kind: 'N', value: 'B' };
+ * const notifE = { kind: 'E', error: new TypeError('x.toUpperCase is not a function') }
+ *
  * const materialized = of(notifA, notifB, notifE);
+ *
  * const upperCase = materialized.pipe(dematerialize());
- * upperCase.subscribe(x => console.log(x), e => console.error(e));
+ * upperCase.subscribe({
+ *    next: x => console.log(x),
+ *    error: e => console.error(e)
+ * });
  *
  * // Results in:
  * // A
  * // B
  * // TypeError: x.toUpperCase is not a function
  * ```
- *
- * @see {@link Notification}
  * @see {@link materialize}
  *
  * @return {Observable} An Observable that emits items and notifications
  * embedded in Notification objects emitted by the source Observable.
- * @name dematerialize
  */
-export function dematerialize<T>(): OperatorFunction<Notification<T>, T> {
-  return function dematerializeOperatorFunction(source: Observable<Notification<T>>) {
-    return source.lift(new DeMaterializeOperator());
+export function dematerialize<N extends ObservableNotification<any>>(): OperatorFunction<N, ValueFromNotification<N>> {
+  return function dematerializeOperatorFunction(source: Observable<N>) {
+    return source.lift(new DeMaterializeOperator<N>());
   };
 }
 
-class DeMaterializeOperator<T extends Notification<any>, R> implements Operator<T, R> {
+class DeMaterializeOperator<N extends ObservableNotification<any>> implements Operator<N, ValueFromNotification<N>> {
   call(subscriber: Subscriber<any>, source: any): any {
-    return source.subscribe(new DeMaterializeSubscriber(subscriber));
+    return source.subscribe(new DeMaterializeSubscriber<N>(subscriber));
   }
 }
 
@@ -66,12 +68,12 @@ class DeMaterializeOperator<T extends Notification<any>, R> implements Operator<
  * @ignore
  * @extends {Ignored}
  */
-class DeMaterializeSubscriber<T extends Notification<any>> extends Subscriber<T> {
-  constructor(destination: Subscriber<any>) {
+class DeMaterializeSubscriber<N extends ObservableNotification<any>> extends Subscriber<N> {
+  constructor(destination: Subscriber<ValueFromNotification<N>>) {
     super(destination);
   }
 
-  protected _next(value: T) {
-    value.observe(this.destination);
+  protected _next(notification: N) {
+    observeNotification(notification, this.destination);
   }
 }
diff --git a/src/internal/operators/materialize.ts b/src/internal/operators/materialize.ts
index df9e109e67..ec4fa3ee12 100644
--- a/src/internal/operators/materialize.ts
+++ b/src/internal/operators/materialize.ts
@@ -2,7 +2,7 @@ import { Operator } from '../Operator';
 import { Observable } from '../Observable';
 import { Subscriber } from '../Subscriber';
 import { Notification } from '../Notification';
-import { OperatorFunction } from '../types';
+import { OperatorFunction, ObservableNotification } from '../types';
 
 /**
  * Represents all of the notifications from the source Observable as `next`
@@ -27,7 +27,9 @@ import { OperatorFunction } from '../types';
  * {@link dematerialize}.
  *
  * ## Example
+ *
  * Convert a faulty Observable to an Observable of Notifications
+ *
  * ```ts
  * import { of } from 'rxjs';
  * import { materialize, map } from 'rxjs/operators';
@@ -51,16 +53,19 @@ import { OperatorFunction } from '../types';
  * @return {Observable<Notification<T>>} An Observable that emits
  * {@link Notification} objects that wrap the original emissions from the source
  * Observable with metadata.
- * @name materialize
+ *
+ * @deprecated In version 8, materialize will start to emit {@link ObservableNotification} objects, and not
+ * {@link Notification} instances. This means that methods that are not commonly used, like `Notification.observe`
+ * will not be available on the emitted values at that time.
  */
-export function materialize<T>(): OperatorFunction<T, Notification<T>> {
+export function materialize<T>(): OperatorFunction<T, Notification<T> & ObservableNotification<T>> {
   return function materializeOperatorFunction(source: Observable<T>) {
-    return source.lift(new MaterializeOperator());
+    return source.lift(new MaterializeOperator<T>());
   };
 }
 
-class MaterializeOperator<T> implements Operator<T, Notification<T>> {
-  call(subscriber: Subscriber<Notification<T>>, source: any): any {
+class MaterializeOperator<T> implements Operator<T, Notification<T> & ObservableNotification<T>> {
+  call(subscriber: Subscriber<Notification<T> & ObservableNotification<T>>, source: any): any {
     return source.subscribe(new MaterializeSubscriber(subscriber));
   }
 }
diff --git a/src/internal/operators/observeOn.ts b/src/internal/operators/observeOn.ts
index b9d0031fa4..0f8875cd9a 100644
--- a/src/internal/operators/observeOn.ts
+++ b/src/internal/operators/observeOn.ts
@@ -1,9 +1,14 @@
 import { Observable } from '../Observable';
 import { Operator } from '../Operator';
 import { Subscriber } from '../Subscriber';
-import { Subscription } from '../Subscription';
-import { Notification } from '../Notification';
-import { MonoTypeOperatorFunction, PartialObserver, SchedulerAction, SchedulerLike, TeardownLogic } from '../types';
+import { observeNotification, COMPLETE_NOTIFICATION, nextNotification, errorNotification } from '../Notification';
+import {
+  MonoTypeOperatorFunction,
+  SchedulerAction,
+  SchedulerLike,
+  TeardownLogic,
+  ObservableNotification,
+} from '../types';
 
 /**
  *
@@ -32,7 +37,9 @@ import { MonoTypeOperatorFunction, PartialObserver, SchedulerAction, SchedulerLi
  * for notification emissions in general.
  *
  * ## Example
+ *
  * Ensure values in subscribe are called just before browser repaint.
+ *
  * ```ts
  * import { interval, animationFrameScheduler } from 'rxjs';
  * import { observeOn } from 'rxjs/operators';
@@ -54,8 +61,6 @@ import { MonoTypeOperatorFunction, PartialObserver, SchedulerAction, SchedulerLi
  * @param {number} [delay] Number of milliseconds that states with what delay every notification should be rescheduled.
  * @return {Observable<T>} Observable that emits the same notifications as the source Observable,
  * but with provided scheduler.
- *
- * @name observeOn
  */
 export function observeOn<T>(scheduler: SchedulerLike, delay: number = 0): MonoTypeOperatorFunction<T> {
   return function observeOnOperatorFunction(source: Observable<T>): Observable<T> {
@@ -63,60 +68,52 @@ export function observeOn<T>(scheduler: SchedulerLike, delay: number = 0): MonoT
   };
 }
 
-export class ObserveOnOperator<T> implements Operator<T, T> {
-  constructor(private scheduler: SchedulerLike, private delay: number = 0) {
-  }
+class ObserveOnOperator<T> implements Operator<T, T> {
+  constructor(private scheduler: SchedulerLike, private delay: number = 0) {}
 
   call(subscriber: Subscriber<T>, source: any): TeardownLogic {
     return source.subscribe(new ObserveOnSubscriber(subscriber, this.scheduler, this.delay));
   }
 }
 
-/**
- * We need this JSDoc comment for affecting ESDoc.
- * @ignore
- * @extends {Ignored}
- */
-export class ObserveOnSubscriber<T> extends Subscriber<T> {
+class ObserveOnSubscriber<T> extends Subscriber<T> {
   /** @nocollapse */
   static dispatch(this: SchedulerAction<ObserveOnMessage>, arg: ObserveOnMessage) {
     const { notification, destination } = arg;
-    notification.observe(destination);
+    observeNotification(notification, destination);
     this.unsubscribe();
   }
 
-  constructor(destination: Subscriber<T>,
-              private scheduler: SchedulerLike,
-              private delay: number = 0) {
+  constructor(destination: Subscriber<T>, private scheduler: SchedulerLike, private delay: number = 0) {
     super(destination);
   }
 
-  private scheduleMessage(notification: Notification<any>): void {
-    const destination = this.destination as Subscription;
-    destination.add(this.scheduler.schedule(
-      ObserveOnSubscriber.dispatch as any,
-      this.delay,
-      new ObserveOnMessage(notification, this.destination)
-    ));
+  private scheduleMessage(notification: ObservableNotification<any>): void {
+    const destination = this.destination as Subscriber<T>;
+    destination.add(
+      this.scheduler.schedule(ObserveOnSubscriber.dispatch as any, this.delay, {
+        notification,
+        destination,
+      })
+    );
   }
 
   protected _next(value: T): void {
-    this.scheduleMessage(Notification.createNext(value));
+    this.scheduleMessage(nextNotification(value));
   }
 
-  protected _error(err: any): void {
-    this.scheduleMessage(Notification.createError(err));
+  protected _error(error: any): void {
+    this.scheduleMessage(errorNotification(error));
     this.unsubscribe();
   }
 
   protected _complete(): void {
-    this.scheduleMessage(Notification.createComplete());
+    this.scheduleMessage(COMPLETE_NOTIFICATION);
     this.unsubscribe();
   }
 }
 
-export class ObserveOnMessage {
-  constructor(public notification: Notification<any>,
-              public destination: PartialObserver<any>) {
-  }
+interface ObserveOnMessage {
+  notification: ObservableNotification<any>;
+  destination: Subscriber<any>;
 }
diff --git a/src/internal/testing/ColdObservable.ts b/src/internal/testing/ColdObservable.ts
index 7460ca6c02..e8c22d3a5b 100644
--- a/src/internal/testing/ColdObservable.ts
+++ b/src/internal/testing/ColdObservable.ts
@@ -6,6 +6,7 @@ import { SubscriptionLog } from './SubscriptionLog';
 import { SubscriptionLoggable } from './SubscriptionLoggable';
 import { applyMixins } from '../util/applyMixins';
 import { Subscriber } from '../Subscriber';
+import { observeNotification } from '../Notification';
 
 /**
  * We need this JSDoc comment for affecting ESDoc.
@@ -20,15 +21,16 @@ export class ColdObservable<T> extends Observable<T> implements SubscriptionLogg
   // @ts-ignore: Property has no initializer and is not definitely assigned
   logUnsubscribedFrame: (index: number) => void;
 
-  constructor(public messages: TestMessage[],
-              scheduler: Scheduler) {
+  constructor(public messages: TestMessage[], scheduler: Scheduler) {
     super(function (this: Observable<T>, subscriber: Subscriber<any>) {
       const observable: ColdObservable<T> = this as any;
       const index = observable.logSubscribedFrame();
       const subscription = new Subscription();
-      subscription.add(new Subscription(() => {
-        observable.logUnsubscribedFrame(index);
-      }));
+      subscription.add(
+        new Subscription(() => {
+          observable.logUnsubscribedFrame(index);
+        })
+      );
       observable.scheduleMessages(subscriber);
       return subscription;
     });
@@ -41,9 +43,9 @@ export class ColdObservable<T> extends Observable<T> implements SubscriptionLogg
       const message = this.messages[i];
       subscriber.add(
         this.scheduler.schedule(
-          state => {
+          (state) => {
             const { message, subscriber } = state!;
-            message.notification.observe(subscriber);
+            observeNotification(message.notification, subscriber);
           },
           message.frame,
           { message, subscriber }
diff --git a/src/internal/testing/HotObservable.ts b/src/internal/testing/HotObservable.ts
index 62e92649d6..9c2a01d96c 100644
--- a/src/internal/testing/HotObservable.ts
+++ b/src/internal/testing/HotObservable.ts
@@ -6,6 +6,7 @@ import { TestMessage } from './TestMessage';
 import { SubscriptionLog } from './SubscriptionLog';
 import { SubscriptionLoggable } from './SubscriptionLoggable';
 import { applyMixins } from '../util/applyMixins';
+import { observeNotification } from '../Notification';
 
 /**
  * We need this JSDoc comment for affecting ESDoc.
@@ -20,8 +21,7 @@ export class HotObservable<T> extends Subject<T> implements SubscriptionLoggable
   // @ts-ignore: Property has no initializer and is not definitely assigned
   logUnsubscribedFrame: (index: number) => void;
 
-  constructor(public messages: TestMessage[],
-              scheduler: Scheduler) {
+  constructor(public messages: TestMessage[], scheduler: Scheduler) {
     super();
     this.scheduler = scheduler;
   }
@@ -31,9 +31,11 @@ export class HotObservable<T> extends Subject<T> implements SubscriptionLoggable
     const subject: HotObservable<T> = this;
     const index = subject.logSubscribedFrame();
     const subscription = new Subscription();
-    subscription.add(new Subscription(() => {
-      subject.logUnsubscribedFrame(index);
-    }));
+    subscription.add(
+      new Subscription(() => {
+        subject.logUnsubscribedFrame(index);
+      })
+    );
     subscription.add(super._subscribe(subscriber));
     return subscription;
   }
@@ -45,11 +47,10 @@ export class HotObservable<T> extends Subject<T> implements SubscriptionLoggable
     for (var i = 0; i < messagesLength; i++) {
       (() => {
         var message = subject.messages[i];
-   /* tslint:enable */
-        subject.scheduler.schedule(
-          () => { message.notification.observe(subject); },
-          message.frame
-        );
+        /* tslint:enable */
+        subject.scheduler.schedule(() => {
+          observeNotification(message.notification, subject);
+        }, message.frame);
       })();
     }
   }
diff --git a/src/internal/testing/TestMessage.ts b/src/internal/testing/TestMessage.ts
index a95c71c4f9..918b47775b 100644
--- a/src/internal/testing/TestMessage.ts
+++ b/src/internal/testing/TestMessage.ts
@@ -1,7 +1,7 @@
-import { Notification } from '../Notification';
+import { ObservableNotification } from '../types';
 
 export interface TestMessage {
   frame: number;
-  notification: Notification<any>;
+  notification: ObservableNotification<any>;
   isGhost?: boolean;
 }
diff --git a/src/internal/testing/TestScheduler.ts b/src/internal/testing/TestScheduler.ts
index a07f879a37..03c578f3c4 100644
--- a/src/internal/testing/TestScheduler.ts
+++ b/src/internal/testing/TestScheduler.ts
@@ -1,5 +1,4 @@
 import { Observable } from '../Observable';
-import { Notification } from '../Notification';
 import { ColdObservable } from './ColdObservable';
 import { HotObservable } from './HotObservable';
 import { TestMessage } from './TestMessage';
@@ -7,6 +6,8 @@ import { SubscriptionLog } from './SubscriptionLog';
 import { Subscription } from '../Subscription';
 import { VirtualTimeScheduler, VirtualAction } from '../scheduler/VirtualTimeScheduler';
 import { AsyncScheduler } from '../scheduler/AsyncScheduler';
+import { ObservableNotification } from '../types';
+import { COMPLETE_NOTIFICATION, errorNotification, nextNotification } from '../Notification';
 
 const defaultMaxFrame: number = 750;
 
@@ -111,11 +112,11 @@ export class TestScheduler extends VirtualTimeScheduler {
                                      outerFrame: number): TestMessage[] {
     const messages: TestMessage[] = [];
     observable.subscribe((value) => {
-      messages.push({ frame: this.frame - outerFrame, notification: Notification.createNext(value) });
-    }, (err) => {
-      messages.push({ frame: this.frame - outerFrame, notification: Notification.createError(err) });
+      messages.push({ frame: this.frame - outerFrame, notification: nextNotification(value) });
+    }, (error) => {
+      messages.push({ frame: this.frame - outerFrame, notification: errorNotification(error) });
     }, () => {
-      messages.push({ frame: this.frame - outerFrame, notification: Notification.createComplete() });
+      messages.push({ frame: this.frame - outerFrame, notification: COMPLETE_NOTIFICATION });
     });
     return messages;
   }
@@ -137,11 +138,11 @@ export class TestScheduler extends VirtualTimeScheduler {
         if (x instanceof Observable) {
           value = this.materializeInnerObservable(value, this.frame);
         }
-        actual.push({ frame: this.frame, notification: Notification.createNext(value) });
-      }, (err) => {
-        actual.push({ frame: this.frame, notification: Notification.createError(err) });
+        actual.push({ frame: this.frame, notification: nextNotification(value) });
+      }, (error) => {
+        actual.push({ frame: this.frame, notification: errorNotification(error) });
       }, () => {
-        actual.push({ frame: this.frame, notification: Notification.createComplete() });
+        actual.push({ frame: this.frame, notification: COMPLETE_NOTIFICATION });
       });
     }, subscriptionFrame);
 
@@ -321,7 +322,7 @@ export class TestScheduler extends VirtualTimeScheduler {
         nextFrame += count * this.frameTimeFactor;
       };
 
-      let notification: Notification<any> | undefined;
+      let notification: ObservableNotification<any> | undefined;
       const c = marbles[i];
       switch (c) {
         case ' ':
@@ -342,14 +343,14 @@ export class TestScheduler extends VirtualTimeScheduler {
           advanceFrameBy(1);
           break;
         case '|':
-          notification = Notification.createComplete();
+          notification = COMPLETE_NOTIFICATION;
           advanceFrameBy(1);
           break;
         case '^':
           advanceFrameBy(1);
           break;
         case '#':
-          notification = Notification.createError(errorValue || 'error');
+          notification = errorNotification(errorValue || 'error');
           advanceFrameBy(1);
           break;
         default:
@@ -386,7 +387,7 @@ export class TestScheduler extends VirtualTimeScheduler {
             }
           }
 
-          notification = Notification.createNext(getValue(c));
+          notification = nextNotification(getValue(c));
           advanceFrameBy(1);
           break;
       }
diff --git a/src/internal/types.ts b/src/internal/types.ts
index 6163132ae9..8a622dd315 100644
--- a/src/internal/types.ts
+++ b/src/internal/types.ts
@@ -80,6 +80,42 @@ export type ObservableLike<T> = InteropObservable<T>;
 
 export type InteropObservable<T> = { [Symbol.observable]: () => Subscribable<T>; };
 
+/** NOTIFICATIONS */
+
+/**
+ * A notification representing a "next" from an observable.
+ * Can be used with {@link dematerialize}.
+ */
+export interface NextNotification<T> {
+  /** The kind of notification. Always "N" */
+  kind: 'N';
+  /** The value of the notification. */
+  value: T;
+}
+
+/**
+ * A notification representing an "error" from an observable.
+ * Can be used with {@link dematerialize}.
+ */
+export interface ErrorNotification {
+  /** The kind of notification. Always "E" */
+  kind: 'E';
+  error: any;
+}
+
+/**
+ * A notification representing a "completion" from an observable.
+ * Can be used with {@link dematerialize}.
+ */
+export interface CompleteNotification {
+  kind: 'C';
+}
+
+/**
+ * Valid observable notification types.
+ */
+export type ObservableNotification<T> = NextNotification<T> | ErrorNotification | CompleteNotification;
+
 /** OBSERVER INTERFACES */
 
 export interface NextObserver<T> {
@@ -201,4 +237,13 @@ export type Tail<X extends any[]> =
  * If you have `T extends Array<any>`, and pass a `string[]` to it,
  * `ValueFromArray<T>` will return the actual type of `string`.
  */
-export type ValueFromArray<A> = A extends Array<infer T> ? T : never;
\ No newline at end of file
+export type ValueFromArray<A> = A extends Array<infer T> ? T : never;
+
+/**
+ * Gets the value type from an {@link ObservableNotification}, if possible.
+ */
+export type ValueFromNotification<T> = T extends { kind: 'N'|'E'|'C' } ?
+  (T extends NextNotification<any> ?
+    (T extends { value: infer V } ? V : undefined )
+  : never)
+  : never;
\ No newline at end of file