Merge branch 'master' into breaking-changes

Author: mikuso

README.md

@@ -313,12 +313,13 @@ The callback function is called with the following three arguments:
   * `headers` {Object} - The HTTP headers sent in the upgrade request.
   * `request` {http.IncomingMessage} - The full HTTP request received by the underlying webserver.
 
+* `signal` {AbortSignal} - An `AbortSignal` used to indicate whether the websocket upgrade has been aborted during the authentication process.  The `signal.reason` is also available as the `error` property of the ['upgradeAborted'](#event-upgradeaborted) event.
 
 Example:
 
 ```js
 const rpcServer = new RPCServer();
-rpcServer.auth((accept, reject, handshake) => {
+rpcServer.auth((accept, reject, handshake, signal) => {
     if (handshake.identity === 'TEST') {
         accept();
     } else {

lib/server.js

@@ -81,7 +81,9 @@ class RPCServer extends EventEmitter {
         return async (request, socket, head) => {
 
             let resolved = false;
-            let emittedAbort = false;
+
+            const ac = new AbortController();
+            const {signal} = ac;
 
             const url = new URL(request.url, 'http://localhost');
             const pathParts = url.pathname.split('/');
@@ -96,8 +98,8 @@ class RPCServer extends EventEmitter {
                     abortHandshake(socket, 500);
                 }
 
-                if (!emittedAbort) {
-                    emittedAbort = true;
+                if (!signal.aborted) {
+                    ac.abort(error);
                     this.emit('upgradeAborted', {
                         error,
                         socket,
@@ -187,6 +189,10 @@ class RPCServer extends EventEmitter {
                     abortUpgrade(new WebsocketUpgradeError(code, message));
                 };
 
+                socket.once('end', () => {
+                    reject(400, `Client connection closed before upgrade complete`);
+                });
+
                 socket.once('close', () => {
                     reject(400, `Client connection closed before upgrade complete`);
                 });
@@ -199,7 +205,8 @@ class RPCServer extends EventEmitter {
                     await this.authCallback(
                         accept,
                         reject,
-                        handshake
+                        handshake,
+                        signal
                     );
                 } else {
                     accept();

package-lock.json

@@ -1,6 +1,6 @@
 {
   "name": "ocpp-rpc",
-  "version": "1.8.0",
+  "version": "1.9.0",
   "description": "A client & server implementation of the WAMP-like RPC-over-websocket system defined in the OCPP protcols (e.g. OCPP1.6-J and OCPP2.0.1).",
   "main": "index.js",
   "scripts": {

package.json

@@ -937,6 +937,37 @@ describe('RPCServer', function(){
 
         });
 
+        it("should abort auth on upgrade error", async () => {
+
+            const {endpoint, close, server} = await createServer();
+            const cli = new RPCClient({
+                endpoint,
+                identity: 'X',
+            });
+
+            let completeAuth;
+            let authCompleted = new Promise(r => {completeAuth = r;})
+
+            server.auth(async (accept, reject, handshake, signal) => {
+                const abortProm = once(signal, 'abort');
+                await cli.close({force: true, awaitPending: false});
+                await abortProm;
+                completeAuth();
+            });
+
+            try {
+                const connErr = await cli.connect().catch(e=>e);
+                await authCompleted;
+                assert.ok(connErr instanceof Error);
+
+            } finally {
+                await cli.close();
+                close();
+            }
+
+        });
+
+
     });
 
 });