Fix Subscriber model when removed within emit callback.

src.ts/providers/abstract-provider.ts

@@ -161,6 +161,8 @@ type Sub = {
     nameMap: Map<string, string>
     addressableMap: WeakMap<Addressable, string>;
     listeners: Array<{ listener: Listener, once: boolean }>;
+    // @TODO: get rid of this, as it is (and has to be)
+    // tracked in subscriber
     started: boolean;
     subscriber: Subscriber;
 };

src.ts/providers/subscriber-connection.ts

@@ -27,15 +27,21 @@ export class BlockConnectionSubscriber implements Subscriber {
     #provider: ConnectionRpcProvider;
     #blockNumber: number;
 
+    #running: boolean;
+
     #filterId: null | number;
 
     constructor(provider: ConnectionRpcProvider) {
         this.#provider = provider;
         this.#blockNumber = -2;
+        this.#running = false;
         this.#filterId = null;
     }
 
     start(): void {
+        if (this.#running) { return; }
+        this.#running = true;
+
         this.#filterId = this.#provider._subscribe([ "newHeads" ], (result: any) => {
             const blockNumber = getNumber(result.number);
             const initial = (this.#blockNumber === -2) ? blockNumber: (this.#blockNumber + 1)
@@ -47,6 +53,9 @@ export class BlockConnectionSubscriber implements Subscriber {
     }
 
     stop(): void {
+        if (!this.#running) { return; }
+        this.#running = false;
+
         if (this.#filterId != null) {
             this.#provider._unsubscribe(this.#filterId);
             this.#filterId = null;

src.ts/providers/subscriber-filterid.ts

@@ -26,6 +26,8 @@ export class FilterIdSubscriber implements Subscriber {
     #filterIdPromise: null | Promise<string>;
     #poller: (b: number) => Promise<void>;
 
+    #running: boolean;
+
     #network: null | Network;
 
     #hault: boolean;
@@ -36,6 +38,8 @@ export class FilterIdSubscriber implements Subscriber {
         this.#filterIdPromise = null;
         this.#poller = this.#poll.bind(this);
 
+        this.#running = false;
+
         this.#network = null;
 
         this.#hault = false;
@@ -91,9 +95,17 @@ export class FilterIdSubscriber implements Subscriber {
         }
     }
 
-    start(): void { this.#poll(-2); }
+    start(): void {
+        if (this.#running) { return; }
+        this.#running = true;
+
+        this.#poll(-2);
+    }
 
     stop(): void {
+        if (!this.#running) { return; }
+        this.#running = false;
+
         this.#hault = true;
         this.#teardown();
         this.#provider.off("block", this.#poller);

src.ts/providers/subscriber-polling.ts

@@ -76,13 +76,13 @@ export class PollingBlockSubscriber implements Subscriber {
     }
 
     start(): void {
-        if (this.#poller) { throw new Error("subscriber already running"); }
+        if (this.#poller) { return; }
         this.#poller = this.#provider._setTimeout(this.#poll.bind(this), this.#interval);
         this.#poll();
     }
 
     stop(): void {
-        if (!this.#poller) { throw new Error("subscriber not running"); }
+        if (!this.#poller) { return; }
         this.#provider._clearTimeout(this.#poller);
         this.#poller = null;
     }
@@ -105,9 +105,11 @@ export class PollingBlockSubscriber implements Subscriber {
 export class OnBlockSubscriber implements Subscriber {
     #provider: AbstractProvider;
     #poll: (b: number) => void;
+    #running: boolean;
 
     constructor(provider: AbstractProvider) {
         this.#provider = provider;
+        this.#running = false;
         this.#poll = (blockNumber: number) => {
             this._poll(blockNumber, this.#provider);
         }
@@ -118,11 +120,17 @@ export class OnBlockSubscriber implements Subscriber {
     }
 
     start(): void {
+        if (this.#running) { return; }
+        this.#running = true;
+
         this.#poll(-2);
         this.#provider.on("block", this.#poll);
     }
 
     stop(): void {
+        if (!this.#running) { return; }
+        this.#running = false;
+
         this.#provider.off("block", this.#poll);
     }
 
@@ -178,6 +186,8 @@ export class PollingEventSubscriber implements Subscriber {
     #filter: EventFilter;
     #poller: (b: number) => void;
 
+    #running: boolean;
+
     // The most recent block we have scanned for events. The value -2
     // indicates we still need to fetch an initial block number
     #blockNumber: number;
@@ -186,6 +196,7 @@ export class PollingEventSubscriber implements Subscriber {
         this.#provider = provider;
         this.#filter = copy(filter);
         this.#poller = this.#poll.bind(this);
+        this.#running = false;
         this.#blockNumber = -2;
     }
 
@@ -215,6 +226,9 @@ export class PollingEventSubscriber implements Subscriber {
     }
 
     start(): void {
+        if (this.#running) { return; }
+        this.#running = true;
+
         if (this.#blockNumber === -2) {
             this.#provider.getBlockNumber().then((blockNumber) => {
                 this.#blockNumber = blockNumber;
@@ -224,6 +238,9 @@ export class PollingEventSubscriber implements Subscriber {
     }
 
     stop(): void {
+        if (!this.#running) { return; }
+        this.#running = false;
+
         this.#provider.off("block", this.#poller);
     }