Reverts #772, forcing recreation of event sources when streaming (#780)

* Add test which fails w/o the change
* Revert "Stop re-creating EventSource connections on close events (#772)"
This reverts commit 9f57a3c.

CHANGELOG.md

@@ -5,6 +5,10 @@ A breaking change will get clearly marked in this log.
 
 ## Unreleased
 
+### Fix
+
+- Reverts a change from [v10.0.0](#v10.0.0) which caused streams to die prematurely ([#780](https://github.com/stellar/js-stellar-sdk/pull/780)).
+
 
 ## [v10.1.0](https://github.com/stellar/js-stellar-sdk/compare/v10.0.1...v10.1.0-beta.0)
 

src/call_builder.ts

@@ -137,9 +137,18 @@ export class CallBuilder<
       createTimeout();
 
       if (es) {
+        // when receiving the close message from Horizon we should
+        // close the connection and recreate the event source
+        let closed = false;
         const onClose = () => {
+          if (closed) {
+            return;
+          }
+
           clearTimeout(timeout);
           es.close();
+          createEventSource();
+          closed = true;
         };
 
         const onMessage = (message: any) => {

test/integration/streaming_test.js

@@ -81,3 +81,44 @@ describe("integration tests: streaming", function(done) {
     });
   });
 });
+
+describe("integration tests:live streaming", function(done) {
+  if (typeof window !== "undefined") {
+    done();
+    return;
+  }
+
+  // stream transactions from pubnet for a while and ensure that we cross a
+  // ledger boundary (if streaming is broken, we will get stuck on a single
+  // ledger's transaction batch).
+  it("streams in perpetuity", function(done) {
+    const DURATION = 30;
+    const server = new StellarSdk.Server("https://horizon.stellar.org");
+    this.timeout((DURATION + 5) * 1000); // pad timeout
+
+    let transactions = [];
+
+    const finishTest = (err) => {
+      clearTimeout(timeout);
+      closeHandler();
+
+      expect(transactions.length).to.be.gt(0);
+      let firstLedger = transactions[0].ledger_attr;
+      let lastLedger = transactions[transactions.length - 1].ledger_attr;
+      expect(lastLedger - firstLedger).to.be.gte(1);
+
+      done(err);
+    }
+
+    let closeHandler = server.transactions()
+      .cursor("now")
+      .stream({
+        onmessage: (msg) => {
+          transactions.push(msg);
+        },
+        onerror: finishTest,
+      });
+
+    let timeout = setTimeout(finishTest, DURATION * 1000);
+  });
+});