fix(notifier): add Far/Data to notifier (#2565)

* fix(notifier): extract many fixes from Kate's PR

This pulls in a subset of the changes from #2545 , which seemed easier than
iterating on that branch or applying more fixes on top. Thanks @katelynsills
for the cleanup!

* fix(notifier): add Far to subscriber

refs #2018

Author: warner

packages/notifier/package.json

@@ -37,11 +37,11 @@
   "dependencies": {
     "@agoric/assert": "^0.2.2",
     "@agoric/eventual-send": "^0.13.2",
+    "@agoric/marshal": "^0.3.2",
     "@agoric/promise-kit": "^0.2.2"
   },
   "devDependencies": {
     "@agoric/install-ses": "^0.5.2",
-    "@agoric/marshal": "^0.3.2",
     "ava": "^3.12.1",
     "esm": "^3.2.25",
     "nyc": "^15.1.0"

packages/notifier/src/notifier.js

@@ -4,6 +4,7 @@
 
 import { makePromiseKit } from '@agoric/promise-kit';
 import { assert } from '@agoric/assert';
+import { Far } from '@agoric/marshal';
 import {
   makeAsyncIterableFromNotifier,
   observeIteration,
@@ -68,7 +69,7 @@ export const makeNotifierKit = (...args) => {
     },
   });
 
-  const notifier = harden({
+  const notifier = Far('notifier', {
     ...makeNotifier(baseNotifier),
     // TODO stop exposing baseNotifier methods directly
     ...baseNotifier,

packages/notifier/src/subscriber.js

@@ -3,6 +3,7 @@
 /// <reference types="ses"/>
 
 import { HandledPromise, E } from '@agoric/eventual-send';
+import { Far } from '@agoric/marshal';
 import { makePromiseKit } from '@agoric/promise-kit';
 
 import './types';
@@ -13,7 +14,7 @@ import './types';
  * @returns {Subscription<T>}
  */
 const makeSubscription = startP => {
-  return harden({
+  return Far('Subscription', {
     // eslint-disable-next-line no-use-before-define
     [Symbol.asyncIterator]: () => makeSubscriptionIterator(startP),
 
@@ -39,7 +40,7 @@ export { makeSubscription };
 const makeSubscriptionIterator = tailP => {
   // To understand the implementation, start with
   // https://web.archive.org/web/20160404122250/http://wiki.ecmascript.org/doku.php?id=strawman:concurrency#infinite_queue
-  return harden({
+  return Far('SubscriptionIterator', {
     subscribe: () => makeSubscription(tailP),
     [Symbol.asyncIterator]: () => makeSubscriptionIterator(tailP),
     next: () => {
@@ -61,7 +62,7 @@ const makeSubscriptionKit = () => {
   let rear;
   const subscription = makeSubscription(new HandledPromise(r => (rear = r)));
 
-  const publication = harden({
+  const publication = Far('publication', {
     updateState: value => {
       if (rear === undefined) {
         throw new Error('Cannot update state after termination.');