Terminate active HTTP requests when session closed

All pending HTTP requests will now be aborted and failed when parent
HTTP session is closed.

Author: lutovich

src/v1/internal/http/http-session.js

@@ -28,7 +28,6 @@ export default class HttpSession extends Session {
   constructor(url, authToken, config) {
     super(WRITE, null, null, config);
     this._statementRunner = new HttpStatementRunner(url, authToken);
-    this._outgoingXhrRequests = {};
   }
 
   run(statement, parameters = {}) {
@@ -58,13 +57,7 @@ export default class HttpSession extends Session {
   }
 
   close(callback = (() => null)) {
-    Object.keys(this._outgoingXhrRequests).forEach(id => {
-      const xhrRequest = this._outgoingXhrRequests[id];
-      if (xhrRequest) {
-        xhrRequest.abort();
-      }
-      delete this._outgoingXhrRequests[id];
-    });
+    this._statementRunner.terminateActiveRequests();
     callback();
   }
 }

src/v1/internal/http/http-statement-runner.js

@@ -22,6 +22,7 @@ import StreamObserver from '../stream-observer';
 import Result from '../../result';
 import {EMPTY_CONNECTION_HOLDER} from '../connection-holder';
 import HttpDataConverter from './http-data-converter';
+import {Neo4jError} from '../../error';
 
 export default class HttpStatementRunner {
 
@@ -30,13 +31,21 @@ export default class HttpStatementRunner {
     this._transactionCommitUrl = createTransactionCommitUrl(url);
     this._headers = createHttpHeaders(authToken);
     this._converter = new HttpDataConverter();
+    this._activeXhrRequests = [];
   }
 
   run(statement, parameters) {
     const streamObserver = new StreamObserver();
-    sendPostRequest(statement, parameters, streamObserver, this._transactionCommitUrl, this._headers, this._converter);
+    sendPostRequest(statement, parameters, streamObserver, this);
     return new Result(streamObserver, statement, parameters, this._serverInfoSupplier, EMPTY_CONNECTION_HOLDER);
   }
+
+  terminateActiveRequests() {
+    while (this._activeXhrRequests.length > 0) {
+      const activeXhrRequest = this._activeXhrRequests.pop();
+      activeXhrRequest.abort();
+    }
+  }
 }
 
 function createServerInfoSupplier(url) {
@@ -58,16 +67,22 @@ function createHttpHeaders(authToken) {
   };
 }
 
-function sendPostRequest(statement, parameters, streamObserver, transactionCommitUrl, headers, converter) {
+function sendPostRequest(statement, parameters, streamObserver, statementRunner) {
   try {
-    xhr.post(
-      transactionCommitUrl,
+    const xhrRequest = xhr.post(
+      statementRunner._transactionCommitUrl,
       {
-        headers: headers,
-        body: createStatementJson(statement, parameters, converter)
+        headers: statementRunner._headers,
+        body: createStatementJson(statement, parameters, statementRunner._converter)
       },
-      (error, response) => processPostResponse(error, response, converter, streamObserver)
+      (error, response) => processPostResponse(error, response, statementRunner._converter, streamObserver)
     );
+
+    xhrRequest.addEventListener('abort', () => {
+      streamObserver.onError(new Neo4jError('Query terminated because session has been closed'));
+    }, false);
+
+    statementRunner._activeXhrRequests.push(xhrRequest);
   } catch (e) {
     streamObserver.onError(e);
   }

test/internal/http-driver.test.js

@@ -179,6 +179,27 @@ describe('http driver', () => {
     });
   });
 
+  it('should terminate run when session is closed', done => {
+    const boltSession = boltDriver.session();
+    boltSession.run('CREATE (:Node)').then(() => {
+      const tx = boltSession.beginTransaction();
+      tx.run(`MATCH (n:Node) SET n.name = 'Thor' RETURN n`).then(() => {
+
+        const httpSession = httpDriver.session();
+        httpSession.run(`MATCH (n:Node) SET n.name = 'Loki' RETURN n`).catch(error => {
+          expect(error.name).toEqual('Neo4jError');
+          expect(error.message).toEqual('Query terminated because session has been closed');
+          done();
+        });
+
+        // close http session after a second
+        setTimeout(() => {
+          httpSession.close();
+        }, 1000);
+      });
+    });
+  }, 10000);
+
   function testSendAndReceiveWithReturnQuery(values, done) {
     const query = 'RETURN $value';