Dev rxrpc js http transport merge (#18)

* Adding Http transport

* fixing unnecessary rebuilds

* fix polling empty-string bug, and change polling period to 1sec

* getting the data as JSON instead of string

* handling subscribtion in post function

Co-authored-by: Mohaban, Assaf <[email protected]>

Author: assafkm

build.gradle

@@ -12,4 +12,8 @@ task deleteBuildFiles(type: Delete) {
     delete "$projectDir/dist", "$project.buildDir"
 }
 
+yarn {
+    it.args.addAll provider {["--network-timeout", "600000"]}
+}
+
 clean.dependsOn(deleteBuildFiles)

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

@@ -15,7 +15,7 @@ describe('RxRpc Http Transport test suite', function () {
     let resp: {}
 
     beforeEach(() => {
-        transport = new RxRpcHttpTransport("http://funnyName/");
+        transport = new RxRpcHttpTransport("https://funnyName/");
         clientId = "12345678";
         incomingMessages = [];
         resp = {
@@ -31,10 +31,10 @@ describe('RxRpc Http Transport test suite', function () {
         expect(incomingMessages[0]).toEqual(clientId)
     })
 
-    it('Poll with data as a string', async () => {
+    it('Poll multiple messages with data as JSON', async () => {
         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'] = `${data1}\n${data2}`
+        resp['data'] = JSON.parse(`[${data1},\n${data2}]`)
         mockedAxios.post.mockImplementation(() => Promise.resolve(resp));
 
         transport.connect().subscribe(connection => connection.poll().subscribe(msg => {
@@ -46,9 +46,9 @@ describe('RxRpc Http Transport test suite', function () {
         expect(incomingMessages[1]).toEqual(JSON.parse(data2));
     })
 
-    it('Poll with data as an object', async () => {
+    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)
+        resp['data'] = JSON.parse(`[${data}]`)
         mockedAxios.post.mockImplementation(() => Promise.resolve(resp));
 
         transport.connect().subscribe(connection => connection.poll().subscribe(msg => {

src/lib/rxrpc-http-transport.ts

@@ -1,6 +1,6 @@
 import {RxRpcConnection, RxRpcTransport} from './rxrpc-transport';
 import {interval, Observable, of, Subject, Subscription} from 'rxjs';
-import {filter, map, mergeMap, retry, tap} from "rxjs/operators";
+import {mergeMap, map, retry, filter, tap} from "rxjs/operators";
 import {HttpAttributes} from "./rxrpc-http-attributes";
 import axios, {AxiosResponse} from 'axios';
 import {fromPromise} from "rxjs/internal-compatibility";
@@ -26,7 +26,7 @@ export class RxRpcHttpConnection implements RxRpcConnection {
                 retry(options.pollingRetryCount))
             .subscribe(
                 () => {},
-                err => this.incoming.error(err),
+                () => {},
                 () => this.incoming.complete());
     }
 
@@ -43,23 +43,23 @@ export class RxRpcHttpConnection implements RxRpcConnection {
     }
 
     poll(): Observable<any> {
-        return this.post('polling')
-            .pipe(
-                map(resp => resp.data),
-                filter(data => data !== ""),
-                mergeMap(data => {
-                    if(typeof data === 'string') {
-                        return fromArray(data.split("\n").map(s => JSON.parse(s)));
-                    }
-                    return of(data);
-                }),
-                tap(obj => this.incoming.next(obj)));
+        return this.post('polling');
     }
 
-    post(path: string, msg?: any): Observable<AxiosResponse<string>> {
+    post(path: string, msg?: any): Observable<any> {
         const headers = {};
         headers[HttpAttributes.ClientIdAttribute] = this.clientId;
         return fromPromise(axios.post<string>(`${this.uri}/${path}`, msg, {headers: headers}))
+            .pipe(
+                map(resp => resp.data),
+                mergeMap(data => {
+                    return fromArray(data);
+                }),
+                tap(
+                obj => this.incoming.next(obj),
+                err => this.incoming.error(err)
+                )
+            );
     }
 }