Refactor PersistentStream (no behavior changes). (#1041)

This breaks out a number of changes I made as prep for b/80402781 (Continue
retrying streams for 1 minute (idle delay)).

PersistentStream changes:
* Rather than providing a stream event listener to every call of start(),
  the stream listener is now provided once to the constructor and cannot
  be changed.
* Streams can now be restarted indefinitely, even after a call to stop().
  * PersistentStreamState.Stopped was removed and we just return to
    'Initial' after a stop() call.
  * Added `closeCount` member to PersistentStream in order to avoid
    bleedthrough issues with auth and stream events once stop() has
    been called.
  * Calling stop() now triggers the onClose() event listener, which
    simplifies stream cleanup.
* PersistentStreamState.Auth renamed to 'Starting' to better reflect that
  it encompasses both authentication and opening the stream.

RemoteStore changes:
* Creates streams once and just stop() / start()s them as necessary,
  never recreating them completely.
  * Added networkEnabled flag to track whether the network is
    enabled or not, since we no longer null out the streams.
  * Refactored disableNetwork() / enableNetwork() to remove stream
    re-creation.

Misc:
* Comment improvements including a state diagram on PersistentStream.
* Fixed spec test shutdown to schedule via the AsyncQueue to fix
  sequencing order I ran into.

Author: mikelehen

packages/firestore/src/remote/datastore.ts

@@ -1,3 +1,4 @@
+import { WatchStreamListener, WriteStreamListener } from './persistent_stream';
 /**
  * Copyright 2017 Google Inc.
  *
@@ -54,21 +55,27 @@ export class Datastore {
     private serializer: JsonProtoSerializer
   ) {}
 
-  newPersistentWriteStream(): PersistentWriteStream {
+  newPersistentWriteStream(
+    listener: WriteStreamListener
+  ): PersistentWriteStream {
     return new PersistentWriteStream(
       this.queue,
       this.connection,
       this.credentials,
-      this.serializer
+      this.serializer,
+      listener
     );
   }
 
-  newPersistentWatchStream(): PersistentListenStream {
+  newPersistentWatchStream(
+    listener: WatchStreamListener
+  ): PersistentListenStream {
     return new PersistentListenStream(
       this.queue,
       this.connection,
       this.credentials,
-      this.serializer
+      this.serializer,
+      listener
     );
   }