Merge pull request #3 from BackendStack21/adding-onClientConnectionTerminated

adding onClientConnectionTerminated callback

Author: jkyberneees

README.md

@@ -111,6 +111,15 @@ Called when an http response is received from the source.
 The default behavior is `pump(stream, res)`, which will be disabled if the
 option is specified.
 
+##### onClientConnectionTerminated(res, err, response)
+Called when the client HTTP connection to the proxy server unexpectedly terminates before the downstream service response is sent.  
+```js
+// internal implementation
+if (!res.socket || res.socket.destroyed || res.writableEnded) {
+  return onClientConnectionTerminated(res, err, response)
+}
+```
+
 ##### rewriteRequestHeaders(req, headers)
 Called to rewrite the headers of the request, before them being sent to the downstream server. 
 It must return the new headers object.

demos/gateway.js

@@ -5,6 +5,10 @@ const { proxy } = require('../index')({
 })
 
 const service = require('restana')()
-service.all('/service/*', (req, res) => proxy(req, res, req.url, {}))
+service.all('/service/*', (req, res) => proxy(req, res, req.url, {
+  onClientConnectionTerminated (res, _err, response) {
+    console.log('Client connection unexpectedly terminated:' + req.url)
+  }
+}))
 
 service.start(8080)

demos/service.js

@@ -9,4 +9,8 @@ service.get('/service/get', (req, res) => res.send('Hello World!'))
 
 service.post('/service/post', (req, res) => res.send(req.body))
 
+service.get('/service/long', (req, res) => {
+  setTimeout(() => res.end(), 10000)
+})
+
 service.start(3000)

index.d.ts

@@ -2,7 +2,7 @@ import * as Http from 'http';
 import * as Https from 'https';
 import { Stream } from 'pump';
 
-interface Options {
+type Options = {
   base?: string;
   cacheURLs?: number;
   requests?: {
@@ -14,13 +14,20 @@ interface Options {
   rejectUnauthorized?: boolean;
 }
 
+type ProxyRequestResponse = {
+  statusCode: Number;
+  headers: Http.OutgoingHttpHeaders;
+  stream: Stream;
+}
+
 declare function fastProxy(options?: Options): {
   proxy(
     originReq: Http.IncomingMessage,
     originRes: Http.ServerResponse,
     source: string,
     opts?: {
       base?: string;
+      onClientConnectionTerminated?(res: Http.ServerResponse, err: Error, response: ProxyRequestResponse): void;
       onResponse?(req: Http.IncomingMessage, res: Http.ServerResponse, stream: Stream): void;
       rewriteRequestHeaders?(req: Http.IncomingMessage, headers: Http.IncomingHttpHeaders): Http.IncomingHttpHeaders;
       rewriteHeaders?(headers: Http.OutgoingHttpHeaders): Http.OutgoingHttpHeaders;

index.js

@@ -28,6 +28,7 @@ function fastProxy (opts = {}) {
       opts = opts || {}
       const reqOpts = opts.request || {}
       const onResponse = opts.onResponse
+      const onClientConnectionTerminated = opts.onClientConnectionTerminated || onClientConnectionTerminatedNoOp
       const rewriteHeaders = opts.rewriteHeaders || rewriteHeadersNoOp
       const rewriteRequestHeaders = opts.rewriteRequestHeaders || rewriteRequestHeadersNoOp
 
@@ -88,8 +89,8 @@ function fastProxy (opts = {}) {
         request: reqOpts
       }
       request(reqParams, (err, response) => {
-        if (res.socket.destroyed || res.writableEnded) {
-          return
+        if (!res.socket || res.socket.destroyed || res.writableEnded) {
+          return onClientConnectionTerminated(res, err, response)
         }
 
         if (err) {
@@ -156,6 +157,10 @@ function rewriteHeadersNoOp (headers) {
   return headers
 }
 
+function onClientConnectionTerminatedNoOp (_err, response) {
+
+}
+
 function rewriteRequestHeadersNoOp (req, headers) {
   return headers
 }

test/12.client-connection-terminated.test.js

@@ -0,0 +1,75 @@
+/* global describe, it */
+'use strict'
+
+const http = require('http')
+const { expect } = require('chai')
+let gateway, service, close, proxy
+let clientTerminated = false
+
+describe('Client connection terminated', () => {
+  it('init', async () => {
+    const fastProxy = require('../index')({
+      base: 'http://127.0.0.1:3000'
+    })
+
+    proxy = fastProxy.proxy
+    close = fastProxy.close
+  })
+
+  it('init & start gateway', async () => {
+    // init gateway
+    gateway = require('restana')()
+
+    gateway.all('/service/*', function (req, res) {
+      proxy(req, res, req.url, {
+        onClientConnectionTerminated (res, _err, response) {
+          clientTerminated = true
+        }
+      })
+    })
+
+    await gateway.start(8080)
+  })
+
+  it('init & start remote service', async () => {
+    // init remote service
+    service = require('restana')()
+
+    service.get('/service/long', (req, res) => {
+      setTimeout(() => res.end(), 500)
+    })
+
+    await service.start(3000)
+  })
+
+  it('should invoke onClientConnectionTerminated callback', async () => {
+    const options = {
+      host: 'localhost',
+      path: '/service/long',
+      port: 8080,
+      method: 'GET',
+      headers: {
+        'Content-Length': 0
+      }
+    }
+    const req = http.request(options)
+    req.on('error', () => {})
+    req.end()
+
+    await sleep(100)
+    req.destroy()
+
+    await sleep(1000)
+    expect(clientTerminated).to.equal(true)
+  })
+
+  it('close all', async () => {
+    close()
+    await gateway.close()
+    await service.close()
+  })
+})
+
+function sleep (ms) {
+  return new Promise(resolve => setTimeout(resolve, ms))
+}