Close open sessions when HTTP driver is closed

This will rollback all open transactions and cancel running queries.

Author: lutovich

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

@@ -19,30 +19,20 @@
 
 import Driver from '../../driver';
 import HttpSession from './http-session';
+import HttpSessionTracker from './http-session-tracker';
 
 export default class HttpDriver extends Driver {
 
   constructor(url, userAgent, token, config) {
     super(url, userAgent, token, config);
-    this._sessionIdGenerator = 0;
-    this._openSessions = {};
+    this._sessionTracker = new HttpSessionTracker();
   }
 
   session() {
-    const id = this._sessionIdGenerator;
-    this._sessionIdGenerator++;
-    const session = new HttpSession(this._url, this._token, this._config);
-    this._openSessions[id] = session;
-    return session;
+    return new HttpSession(this._url, this._token, this._config, this._sessionTracker);
   }
 
   close() {
-    Object.keys(this._openSessions).forEach(id => {
-      const session = this._openSessions[id];
-      if (session) {
-        session.close();
-      }
-      delete this._openSessions[id];
-    });
+    return this._sessionTracker.close();
   }
 }

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

@@ -0,0 +1,63 @@
+/**
+ * Copyright (c) 2002-2018 "Neo Technology,"
+ * Network Engine for Objects in Lund AB [http://neotechnology.com]
+ *
+ * This file is part of Neo4j.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+export default class HttpSessionTracker {
+
+  constructor() {
+    this._openSessions = new Set();
+  }
+
+  /**
+   * Record given session as open.
+   * @param {HttpSession} session the newly open session.
+   */
+  sessionOpened(session) {
+    this._openSessions.add(session);
+  }
+
+  /**
+   * Record given session as close.
+   * @param {HttpSession} session the just closed session.
+   */
+  sessionClosed(session) {
+    this._openSessions.delete(session);
+  }
+
+  /**
+   * Close this tracker and all open sessions.
+   */
+  close() {
+    const sessions = Array.from(this._openSessions);
+    this._openSessions.clear();
+    return Promise.all(sessions.map(session => closeSession(session)));
+  }
+}
+
+/**
+ * Close given session and get a promise back.
+ * @param {HttpSession} session the session to close.
+ * @return {Promise<void>} promise resolved when session is closed.
+ */
+function closeSession(session) {
+  return new Promise(resolve => {
+    session.close(() => {
+      resolve();
+    });
+  });
+}

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

@@ -27,11 +27,13 @@ import Result from '../../result';
 
 export default class HttpSession extends Session {
 
-  constructor(url, authToken, config) {
+  constructor(url, authToken, config, sessionTracker) {
     super(WRITE, null, null, config);
     this._ongoingTransactionIds = [];
     this._serverInfoSupplier = createServerInfoSupplier(url);
     this._requestRunner = new HttpRequestRunner(url, authToken);
+    this._sessionTracker = sessionTracker;
+    this._sessionTracker.sessionOpened(this);
   }
 
   run(statement, parameters = {}) {
@@ -79,7 +81,11 @@ export default class HttpSession extends Session {
       rollbackTransactionSilently(transactionId, this._requestRunner)
     );
 
-    Promise.all(rollbackAllOngoingTransactions).then(() => callback());
+    Promise.all(rollbackAllOngoingTransactions)
+      .then(() => {
+        this._sessionTracker.sessionClosed(this);
+        callback();
+      });
   }
 }
 

test/internal/http/http-driver.test.js

@@ -463,6 +463,28 @@ describe('http driver', () => {
       done);
   });
 
+  it('should close all open sessions when closed', done => {
+    if (testUtils.isServer()) {
+      done();
+      return;
+    }
+
+    const session1 = withFakeClose(httpDriver.session());
+    const session2 = withFakeClose(httpDriver.session());
+    const session3 = withFakeClose(httpDriver.session());
+
+    expect(session1.closed).toBeFalsy();
+    expect(session2.closed).toBeFalsy();
+    expect(session3.closed).toBeFalsy();
+
+    httpDriver.close().then(() => {
+      expect(session1.closed).toBeTruthy();
+      expect(session2.closed).toBeTruthy();
+      expect(session3.closed).toBeTruthy();
+      done();
+    });
+  });
+
   function testReceiveSingleValueWithHttpDriver(query, expectedValue, done) {
     runQueryAndGetResults(query, {}, httpDriver).then(results => {
       const receivedValue = results[0][0];
@@ -558,4 +580,14 @@ describe('http driver', () => {
     return serverVersion.compareTo(VERSION_3_4_0) >= 0;
   }
 
+  function withFakeClose(httpSession) {
+    httpSession.closed = false;
+    const originalClose = httpSession.close.bind(httpSession);
+    httpSession.close = callback => {
+      httpSession.closed = true;
+      originalClose(callback);
+    };
+    return httpSession;
+  }
+
 });

test/internal/http/http-session-tracker.test.js

@@ -0,0 +1,83 @@
+/**
+ * Copyright (c) 2002-2018 "Neo Technology,"
+ * Network Engine for Objects in Lund AB [http://neotechnology.com]
+ *
+ * This file is part of Neo4j.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+import sharedNeo4j from '../../internal/shared-neo4j';
+import HttpSession from '../../../src/v1/internal/http/http-session';
+import urlUtil from '../../../src/v1/internal/url-util';
+import HttpSessionTracker from '../../../src/v1/internal/http/http-session-tracker';
+
+describe('http session tracker', () => {
+
+  it('should close open sessions', done => {
+    const tracker = new HttpSessionTracker();
+
+    const session1 = new FakeHttpSession(tracker);
+    const session2 = new FakeHttpSession(tracker);
+    const session3 = new FakeHttpSession(tracker);
+
+    tracker.sessionOpened(session1);
+    tracker.sessionOpened(session2);
+    tracker.sessionOpened(session3);
+
+    tracker.close().then(() => {
+      expect(session1.timesClosed).toEqual(1);
+      expect(session2.timesClosed).toEqual(1);
+      expect(session3.timesClosed).toEqual(1);
+      done();
+    });
+  });
+
+  it('should not close closed sessions', done => {
+    const tracker = new HttpSessionTracker();
+
+    const session1 = new FakeHttpSession(tracker);
+    const session2 = new FakeHttpSession(tracker);
+    const session3 = new FakeHttpSession(tracker);
+    const session4 = new FakeHttpSession(tracker);
+
+    tracker.sessionOpened(session1);
+    tracker.sessionOpened(session2);
+    tracker.sessionOpened(session3);
+    tracker.sessionOpened(session4);
+
+    tracker.sessionClosed(session2);
+    tracker.sessionClosed(session4);
+
+    tracker.close().then(() => {
+      expect(session1.timesClosed).toEqual(1);
+      expect(session2.timesClosed).toEqual(0);
+      expect(session3.timesClosed).toEqual(1);
+      expect(session4.timesClosed).toEqual(0);
+      done();
+    });
+  });
+
+});
+
+class FakeHttpSession extends HttpSession {
+
+  constructor(sessionTracker) {
+    super(urlUtil.parseDatabaseUrl('http://localhost:7474'), sharedNeo4j.authToken, {}, sessionTracker);
+    this.timesClosed = 0;
+  }
+
+  close(callback) {
+    this.timesClosed++;
+    callback();
+  }
+}