add interceptors for http transport options (#20)

* add intercptors in http options
* add interceptors headers to connect + fix to close + fix tests
* fix typo

Signed-off-by: Koaz <[email protected]>

Author: urikoa

src/lib/rxrpc-http-transport.spec.ts

@@ -1,8 +1,14 @@
-import {RxRpcHttpConnection, RxRpcHttpTransport} from './rxrpc-http-transport';
+import {
+    RxRpcHttpConnection,
+    RxRpcHttpTransport,
+    RxRpcHttpTransportInterceptor,
+    RxRpcHttpTransportRequestConfig
+} from './rxrpc-http-transport';
 import axios from 'axios';
+import {Observable, of} from "rxjs";
+import {HttpAttributes} from "./rxrpc-http-attributes";
 
 jest.mock('axios');
-const mockedAxios = axios as jest.Mocked<typeof axios>
 
 function delay(ms: number){
     return new Promise( resolve => setTimeout(resolve, ms) );
@@ -14,18 +20,44 @@ describe('RxRpc Http Transport test suite', function () {
     let incomingMessages: any[];
     let resp: {}
 
+    let interceptors: RxRpcHttpTransportInterceptor[] = [];
+    let headerKey1 = "hk1";
+    let headerKey2 = "hk2"
+    let headerValue1 = "value1";
+    let headerValue2 = "value2";
+    let mockedAxios: jest.Mocked<typeof axios>
+
     beforeEach(() => {
-        transport = new RxRpcHttpTransport("https://funnyName/");
+        mockedAxios = axios as jest.Mocked<typeof axios>
+        interceptors.push(new class implements RxRpcHttpTransportInterceptor {
+            intercept(requestConfig: RxRpcHttpTransportRequestConfig): Observable<RxRpcHttpTransportRequestConfig> {
+                requestConfig.headers[headerKey1] = headerValue1;
+                return of(requestConfig);
+            }
+        })
+        interceptors.push(new class implements RxRpcHttpTransportInterceptor {
+            intercept(requestConfig: RxRpcHttpTransportRequestConfig): Observable<RxRpcHttpTransportRequestConfig> {
+                requestConfig.headers[headerKey2] = headerValue2;
+                return of(requestConfig);
+            }
+        })
+
+        transport = new RxRpcHttpTransport("https://funnyName/", {interceptors: interceptors});
+
         clientId = "12345678";
         incomingMessages = [];
         resp = {
             headers: {"x-rpc-client-id": clientId}
         };
     });
 
-    it('Connect', async () => {
-        mockedAxios.post.mockImplementation(() => Promise.resolve(resp));
+    afterEach(() => {
+        mockedAxios.post.mockReset();
+        interceptors = [];
+    });
 
+    it('Connect', async () => {
+        mockAndVerifyExpectedHeaders();
         transport.connect().subscribe(connection => incomingMessages.push(connection['clientId']))
         await delay(1000);
         expect(incomingMessages[0]).toEqual(clientId)
@@ -35,11 +67,13 @@ describe('RxRpc Http Transport test suite', function () {
         const data1 = "{\"invocationId\":1,\"result\":{\"type\":\"Data\",\"data\":\"Hello, Angular #0\",\"error\":null}}"
         const data2 = "{\"invocationId\":1,\"result\":{\"type\":\"Data\",\"data\":\"Hello, Angular #1\",\"error\":null}}"
         resp['data'] = JSON.parse(`[${data1},\n${data2}]`)
-        mockedAxios.post.mockImplementation(() => Promise.resolve(resp));
-
-        transport.connect().subscribe(connection => connection.poll().subscribe(msg => {
-            incomingMessages.push(msg);
-        }))
+        mockAndVerifyExpectedHeaders();
+        transport.connect().subscribe(connection => {
+            mockAndVerifyExpectedHeadersWithClientId();
+            connection.poll().subscribe(msg => {
+                incomingMessages.push(msg);
+            })
+        })
         await delay(1000);
         expect(incomingMessages.length).toEqual(2);
         expect(incomingMessages[0]).toEqual(JSON.parse(data1));
@@ -49,33 +83,61 @@ describe('RxRpc Http Transport test suite', function () {
     it('Poll with data as JSON', async () => {
         const data = "{\"invocationId\":1,\"result\":{\"type\":\"Data\",\"data\":\"Hello, Angular #0\",\"error\":null}}"
         resp['data'] = JSON.parse(`[${data}]`)
-        mockedAxios.post.mockImplementation(() => Promise.resolve(resp));
-
-        transport.connect().subscribe(connection => connection.poll().subscribe(msg => {
-            incomingMessages.push(msg);
-        }))
+        mockAndVerifyExpectedHeaders();
+        transport.connect().subscribe(connection => {
+            mockAndVerifyExpectedHeadersWithClientId();
+            connection.poll().subscribe(msg => {
+                incomingMessages.push(msg);
+            })
+        })
         await delay(1000);
         expect(incomingMessages.length).toEqual(1);
         expect(incomingMessages[0]).toEqual(JSON.parse(data));
     })
 
     it('Close', async () => {
-        mockedAxios.post.mockImplementation(() => Promise.resolve(resp));
-
+        mockAndVerifyExpectedHeaders();
         transport.connect().subscribe(connection => incomingMessages.push(connection))
         await delay(1000);
         const connection =  incomingMessages[0] as RxRpcHttpConnection;
+        mockAndVerifyExpectedHeadersWithClientId();
         connection.close()
         expect(connection['pollingSubscription'].closed).toEqual(true)
     })
 
     it('Error', async () => {
-        mockedAxios.post.mockImplementation(() => Promise.resolve(resp));
-
+        mockAndVerifyExpectedHeaders();
         transport.connect().subscribe(connection => incomingMessages.push(connection))
         await delay(1000);
         const connection =  incomingMessages[0] as RxRpcHttpConnection;
+        mockAndVerifyExpectedHeadersWithClientId();
         connection.error(null)
         expect(connection['pollingSubscription'].closed).toEqual(true)
     })
+
+    function mockAndVerifyExpectedHeaders()   {
+        mockAndVerifyHeaders(getHeaders());
+    }
+    function mockAndVerifyExpectedHeadersWithClientId()   {
+        mockAndVerifyHeaders(getHeadersWithClientId());
+    }
+    function mockAndVerifyHeaders(expectedHeaders: {}) {
+        mockedAxios.post.mockImplementation((url, msg, options) => {
+            expect(options.headers).toEqual(expectedHeaders);
+            return Promise.resolve(resp);
+        });
+    }
+
+    function getHeadersWithClientId() {
+        let expectedHeaders = getHeaders();
+        expectedHeaders[HttpAttributes.ClientIdAttribute] = clientId;
+        return expectedHeaders;
+    }
+
+    function getHeaders() {
+        let expectedHeaders = {};
+        expectedHeaders[headerKey1] = headerValue1;
+        expectedHeaders[headerKey2] = headerValue2;
+        return expectedHeaders;
+    }
 })

src/lib/rxrpc-http-transport.ts

@@ -1,24 +1,43 @@
 import {RxRpcConnection, RxRpcTransport} from './rxrpc-transport';
 import {interval, Observable, of, Subject, Subscription} from 'rxjs';
-import {mergeMap, map, retry, filter, tap} from "rxjs/operators";
+import {map, mergeMap, retry, tap} from "rxjs/operators";
 import {HttpAttributes} from "./rxrpc-http-attributes";
-import axios, {AxiosResponse} from 'axios';
+import axios from 'axios';
 import {fromPromise} from "rxjs/internal-compatibility";
 import {fromArray} from "rxjs/internal/observable/fromArray";
 
 export interface RxRpcHttpTransportOptions {
     pollingPeriodMillis?: number
     pollingRetryCount?: number
+    interceptors?: RxRpcHttpTransportInterceptor[]
+}
+
+export interface RxRpcHttpTransportRequestConfig {
+    headers: {[key: string]: string}
+}
+
+export interface RxRpcHttpTransportInterceptor {
+    intercept(requestConfig: RxRpcHttpTransportRequestConfig) : Observable<RxRpcHttpTransportRequestConfig>;
 }
 
 export class RxRpcHttpConnection implements RxRpcConnection {
+    readonly interceptors: RxRpcHttpTransportInterceptor[] = [];
     readonly messages: Observable<any>;
     private pollingSubscription: Subscription;
     private readonly incoming = new Subject<any>();
 
     constructor(private readonly uri: string,
                 private readonly clientId: string,
                 options: RxRpcHttpTransportOptions) {
+
+        this.interceptors = options.interceptors || [];
+        this.interceptors.push({
+                intercept(requestConfig: RxRpcHttpTransportRequestConfig): Observable<RxRpcHttpTransportRequestConfig> {
+                    requestConfig.headers[HttpAttributes.ClientIdAttribute] = clientId;
+                    return of(requestConfig);
+                }
+        });
+
         this.messages = this.incoming;
         this.pollingSubscription = interval(options.pollingPeriodMillis)
             .pipe(
@@ -32,6 +51,7 @@ export class RxRpcHttpConnection implements RxRpcConnection {
 
     close() {
         this.pollingSubscription.unsubscribe();
+        RxRpcHttpConnection.postWithInterceptors(`${this.uri}/disconnect`, this.interceptors).subscribe();
     }
 
     error(error: any) {
@@ -46,10 +66,20 @@ export class RxRpcHttpConnection implements RxRpcConnection {
         return this.post('polling');
     }
 
-    post(path: string, msg?: any): Observable<any> {
+    static postWithInterceptors(url: string, interceptors: RxRpcHttpTransportInterceptor[], body?: any): Observable<any> {
         const headers = {};
-        headers[HttpAttributes.ClientIdAttribute] = this.clientId;
-        return fromPromise(axios.post<string>(`${this.uri}/${path}`, msg, {headers: headers}))
+
+        let config: Observable<RxRpcHttpTransportRequestConfig> = of({headers: headers});
+
+        // flatten all interceptor observable
+        interceptors.forEach(interceptor => config = config.pipe(mergeMap(cfg => interceptor.intercept(cfg))));
+
+        return config.pipe(mergeMap(cfg =>
+            fromPromise(axios.post<string>(url, body, {headers: cfg.headers}))));
+    }
+
+    post(path: string, msg?: any): Observable<any> {
+        return RxRpcHttpConnection.postWithInterceptors(`${this.uri}/${path}`, this.interceptors, msg)
             .pipe(
                 map(resp => resp.data),
                 mergeMap(data => {
@@ -59,27 +89,27 @@ export class RxRpcHttpConnection implements RxRpcConnection {
                 obj => this.incoming.next(obj),
                 err => this.incoming.error(err)
                 )
-            );
+            )
     }
 }
 
 export class RxRpcHttpTransport implements RxRpcTransport {
     private readonly options: RxRpcHttpTransportOptions;
     private static readonly defaultOptions: RxRpcHttpTransportOptions = {
         pollingPeriodMillis: HttpAttributes.DefaultClientPollingPeriodMillis,
-        pollingRetryCount: HttpAttributes.DefaultClientPollingRetryCount
+        pollingRetryCount: HttpAttributes.DefaultClientPollingRetryCount,
+        interceptors: []
     }
 
     constructor(private readonly uri: string, options?: RxRpcHttpTransportOptions) {
         this.options = {...RxRpcHttpTransport.defaultOptions, ...options} || RxRpcHttpTransport.defaultOptions
     }
 
     connect(): Observable<RxRpcHttpConnection> {
-        return fromPromise(axios.post<string>(`${this.uri}/connect`))
-            .pipe(
-                map( res => {
-                    const clientId = res.headers[HttpAttributes.ClientIdAttribute.toLowerCase()];
-                    return new RxRpcHttpConnection(this.uri, clientId, this.options);
-                }));
+        return RxRpcHttpConnection.postWithInterceptors(`${this.uri}/connect`, this.options.interceptors)
+            .pipe(map( res => {
+                const clientId = res.headers[HttpAttributes.ClientIdAttribute.toLowerCase()];
+                return new RxRpcHttpConnection(this.uri, clientId, this.options);
+            }));
     }
 }