[NP] KibanaRequest provides request abortion event (#55061) (#55538)

* add aborted$ observable to KibanaRequest

* complete observable on request end

* update docs

* update test suit names

* always finish subscription

* address comments

docs/development/core/server/kibana-plugin-server.kibanarequest.events.md

@@ -0,0 +1,13 @@
+<!-- Do not edit this file. It is automatically generated by API Documenter. -->
+
+[Home](./index.md) &gt; [kibana-plugin-server](./kibana-plugin-server.md) &gt; [KibanaRequest](./kibana-plugin-server.kibanarequest.md) &gt; [events](./kibana-plugin-server.kibanarequest.events.md)
+
+## KibanaRequest.events property
+
+Request events [KibanaRequestEvents](./kibana-plugin-server.kibanarequestevents.md)
+
+<b>Signature:</b>
+
+```typescript
+readonly events: KibanaRequestEvents;
+```

docs/development/core/server/kibana-plugin-server.kibanarequest.md

@@ -23,10 +23,11 @@ export declare class KibanaRequest<Params = unknown, Query = unknown, Body = unk
 |  Property | Modifiers | Type | Description |
 |  --- | --- | --- | --- |
 |  [body](./kibana-plugin-server.kibanarequest.body.md) |  | <code>Body</code> |  |
+|  [events](./kibana-plugin-server.kibanarequest.events.md) |  | <code>KibanaRequestEvents</code> | Request events [KibanaRequestEvents](./kibana-plugin-server.kibanarequestevents.md) |
 |  [headers](./kibana-plugin-server.kibanarequest.headers.md) |  | <code>Headers</code> | Readonly copy of incoming request headers. |
 |  [params](./kibana-plugin-server.kibanarequest.params.md) |  | <code>Params</code> |  |
 |  [query](./kibana-plugin-server.kibanarequest.query.md) |  | <code>Query</code> |  |
 |  [route](./kibana-plugin-server.kibanarequest.route.md) |  | <code>RecursiveReadonly&lt;KibanaRequestRoute&lt;Method&gt;&gt;</code> | matched route details |
-|  [socket](./kibana-plugin-server.kibanarequest.socket.md) |  | <code>IKibanaSocket</code> |  |
+|  [socket](./kibana-plugin-server.kibanarequest.socket.md) |  | <code>IKibanaSocket</code> | [IKibanaSocket](./kibana-plugin-server.ikibanasocket.md) |
 |  [url](./kibana-plugin-server.kibanarequest.url.md) |  | <code>Url</code> | a WHATWG URL standard object. |
 

docs/development/core/server/kibana-plugin-server.kibanarequest.socket.md

@@ -4,6 +4,8 @@
 
 ## KibanaRequest.socket property
 
+[IKibanaSocket](./kibana-plugin-server.ikibanasocket.md)
+
 <b>Signature:</b>
 
 ```typescript

docs/development/core/server/kibana-plugin-server.kibanarequestevents.aborted_.md

@@ -0,0 +1,13 @@
+<!-- Do not edit this file. It is automatically generated by API Documenter. -->
+
+[Home](./index.md) &gt; [kibana-plugin-server](./kibana-plugin-server.md) &gt; [KibanaRequestEvents](./kibana-plugin-server.kibanarequestevents.md) &gt; [aborted$](./kibana-plugin-server.kibanarequestevents.aborted_.md)
+
+## KibanaRequestEvents.aborted$ property
+
+Observable that emits once if and when the request has been aborted.
+
+<b>Signature:</b>
+
+```typescript
+aborted$: Observable<void>;
+```

docs/development/core/server/kibana-plugin-server.kibanarequestevents.md

@@ -0,0 +1,20 @@
+<!-- Do not edit this file. It is automatically generated by API Documenter. -->
+
+[Home](./index.md) &gt; [kibana-plugin-server](./kibana-plugin-server.md) &gt; [KibanaRequestEvents](./kibana-plugin-server.kibanarequestevents.md)
+
+## KibanaRequestEvents interface
+
+Request events.
+
+<b>Signature:</b>
+
+```typescript
+export interface KibanaRequestEvents 
+```
+
+## Properties
+
+|  Property | Type | Description |
+|  --- | --- | --- |
+|  [aborted$](./kibana-plugin-server.kibanarequestevents.aborted_.md) | <code>Observable&lt;void&gt;</code> | Observable that emits once if and when the request has been aborted. |
+

docs/development/core/server/kibana-plugin-server.md

@@ -76,6 +76,7 @@ The plugin integrates with the core system via lifecycle events: `setup`<!-- -->
 |  [IRouter](./kibana-plugin-server.irouter.md) | Registers route handlers for specified resource path and method. See [RouteConfig](./kibana-plugin-server.routeconfig.md) and [RequestHandler](./kibana-plugin-server.requesthandler.md) for more information about arguments to route registrations. |
 |  [IScopedRenderingClient](./kibana-plugin-server.iscopedrenderingclient.md) |  |
 |  [IUiSettingsClient](./kibana-plugin-server.iuisettingsclient.md) | Server-side client that provides access to the advanced settings stored in elasticsearch. The settings provide control over the behavior of the Kibana application. For example, a user can specify how to display numeric or date fields. Users can adjust the settings via Management UI. |
+|  [KibanaRequestEvents](./kibana-plugin-server.kibanarequestevents.md) | Request events. |
 |  [KibanaRequestRoute](./kibana-plugin-server.kibanarequestroute.md) | Request specific route information exposed to a handler. |
 |  [LegacyRequest](./kibana-plugin-server.legacyrequest.md) |  |
 |  [LegacyServiceSetupDeps](./kibana-plugin-server.legacyservicesetupdeps.md) |  |

src/core/server/http/index.ts

@@ -29,6 +29,7 @@ export {
   HttpResponsePayload,
   ErrorHttpResponseOptions,
   KibanaRequest,
+  KibanaRequestEvents,
   KibanaRequestRoute,
   KibanaRequestRouteOptions,
   IKibanaResponse,

src/core/server/http/integration_tests/request.test.ts

@@ -0,0 +1,127 @@
+/*
+ * Licensed to Elasticsearch B.V. under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch B.V. licenses this file to you 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 supertest from 'supertest';
+
+import { HttpService } from '../http_service';
+
+import { contextServiceMock } from '../../context/context_service.mock';
+import { loggingServiceMock } from '../../logging/logging_service.mock';
+import { createHttpServer } from '../test_utils';
+
+let server: HttpService;
+
+let logger: ReturnType<typeof loggingServiceMock.create>;
+const contextSetup = contextServiceMock.createSetupContract();
+
+const setupDeps = {
+  context: contextSetup,
+};
+
+beforeEach(() => {
+  logger = loggingServiceMock.create();
+
+  server = createHttpServer({ logger });
+});
+
+afterEach(async () => {
+  await server.stop();
+});
+
+const delay = (ms: number) => new Promise(resolve => setTimeout(resolve, ms));
+describe('KibanaRequest', () => {
+  describe('events', () => {
+    describe('aborted$', () => {
+      it('emits once and completes when request aborted', async done => {
+        expect.assertions(1);
+        const { server: innerServer, createRouter } = await server.setup(setupDeps);
+        const router = createRouter('/');
+
+        const nextSpy = jest.fn();
+        router.get({ path: '/', validate: false }, async (context, request, res) => {
+          request.events.aborted$.subscribe({
+            next: nextSpy,
+            complete: () => {
+              expect(nextSpy).toHaveBeenCalledTimes(1);
+              done();
+            },
+          });
+
+          // prevents the server to respond
+          await delay(30000);
+          return res.ok({ body: 'ok' });
+        });
+
+        await server.start();
+
+        const incomingRequest = supertest(innerServer.listener)
+          .get('/')
+          // end required to send request
+          .end();
+
+        setTimeout(() => incomingRequest.abort(), 50);
+      });
+
+      it('completes & does not emit when request handled', async () => {
+        const { server: innerServer, createRouter } = await server.setup(setupDeps);
+        const router = createRouter('/');
+
+        const nextSpy = jest.fn();
+        const completeSpy = jest.fn();
+        router.get({ path: '/', validate: false }, async (context, request, res) => {
+          request.events.aborted$.subscribe({
+            next: nextSpy,
+            complete: completeSpy,
+          });
+
+          return res.ok({ body: 'ok' });
+        });
+
+        await server.start();
+
+        await supertest(innerServer.listener).get('/');
+
+        expect(nextSpy).toHaveBeenCalledTimes(0);
+        expect(completeSpy).toHaveBeenCalledTimes(1);
+      });
+
+      it('completes & does not emit when request rejected', async () => {
+        const { server: innerServer, createRouter } = await server.setup(setupDeps);
+        const router = createRouter('/');
+
+        const nextSpy = jest.fn();
+        const completeSpy = jest.fn();
+        router.get({ path: '/', validate: false }, async (context, request, res) => {
+          request.events.aborted$.subscribe({
+            next: nextSpy,
+            complete: completeSpy,
+          });
+
+          return res.badRequest();
+        });
+
+        await server.start();
+
+        await supertest(innerServer.listener).get('/');
+
+        expect(nextSpy).toHaveBeenCalledTimes(0);
+        expect(completeSpy).toHaveBeenCalledTimes(1);
+      });
+    });
+  });
+});

src/core/server/http/router/index.ts

@@ -21,6 +21,7 @@ export { Headers, filterHeaders, ResponseHeaders, KnownHeaders } from './headers
 export { Router, RequestHandler, IRouter, RouteRegistrar } from './router';
 export {
   KibanaRequest,
+  KibanaRequestEvents,
   KibanaRequestRoute,
   KibanaRequestRouteOptions,
   isRealRequest,

src/core/server/http/router/request.ts

@@ -19,6 +19,8 @@
 
 import { Url } from 'url';
 import { Request } from 'hapi';
+import { Observable, fromEvent, merge } from 'rxjs';
+import { shareReplay, first, takeUntil } from 'rxjs/operators';
 
 import { deepFreeze, RecursiveReadonly } from '../../../utils';
 import { Headers } from './headers';
@@ -46,6 +48,17 @@ export interface KibanaRequestRoute<Method extends RouteMethod> {
   options: KibanaRequestRouteOptions<Method>;
 }
 
+/**
+ * Request events.
+ * @public
+ * */
+export interface KibanaRequestEvents {
+  /**
+   * Observable that emits once if and when the request has been aborted.
+   */
+  aborted$: Observable<void>;
+}
+
 /**
  * @deprecated
  * `hapi` request object, supported during migration process only for backward compatibility.
@@ -115,7 +128,10 @@ export class KibanaRequest<
    */
   public readonly headers: Headers;
 
+  /** {@link IKibanaSocket} */
   public readonly socket: IKibanaSocket;
+  /** Request events {@link KibanaRequestEvents} */
+  public readonly events: KibanaRequestEvents;
 
   /** @internal */
   protected readonly [requestSymbol]: Request;
@@ -138,12 +154,22 @@ export class KibanaRequest<
       enumerable: false,
     });
 
-    this.route = deepFreeze(this.getRouteInfo());
+    this.route = deepFreeze(this.getRouteInfo(request));
     this.socket = new KibanaSocket(request.raw.req.socket);
+    this.events = this.getEvents(request);
   }
 
-  private getRouteInfo(): KibanaRequestRoute<Method> {
-    const request = this[requestSymbol];
+  private getEvents(request: Request): KibanaRequestEvents {
+    const finish$ = merge(
+      fromEvent(request.raw.req, 'end'), // all data consumed
+      fromEvent(request.raw.req, 'close') // connection was closed
+    ).pipe(shareReplay(1), first());
+    return {
+      aborted$: fromEvent<void>(request.raw.req, 'aborted').pipe(first(), takeUntil(finish$)),
+    } as const;
+  }
+
+  private getRouteInfo(request: Request): KibanaRequestRoute<Method> {
     const method = request.method as Method;
     const { parse, maxBytes, allow, output } = request.route.settings.payload || {};
 

src/core/server/index.ts

@@ -109,6 +109,7 @@ export {
   IKibanaSocket,
   IsAuthenticated,
   KibanaRequest,
+  KibanaRequestEvents,
   KibanaRequestRoute,
   KibanaRequestRouteOptions,
   IKibanaResponse,

src/core/server/server.api.md

@@ -879,6 +879,7 @@ export class KibanaRequest<Params = unknown, Query = unknown, Body = unknown, Me
     constructor(request: Request, params: Params, query: Query, body: Body, withoutSecretHeaders: boolean);
     // (undocumented)
     readonly body: Body;
+    readonly events: KibanaRequestEvents;
     // Warning: (ae-forgotten-export) The symbol "RouteValidator" needs to be exported by the entry point index.d.ts
     //
     // @internal
@@ -894,6 +895,11 @@ export class KibanaRequest<Params = unknown, Query = unknown, Body = unknown, Me
     readonly url: Url;
     }
 
+// @public
+export interface KibanaRequestEvents {
+    aborted$: Observable<void>;
+}
+
 // @public
 export interface KibanaRequestRoute<Method extends RouteMethod> {
     // (undocumented)