Consuming a RESTapi stream with angular2

Question

How is it possible to consume a "never-ending" data stream with angular2? I'm writing a little chat app with a server written in go and a client writting in angular2. For a push service I implemented an endpoint which will keep the connection up. A authorized user can connect to the message broker on server.com:123/broker with a GET request. Every time a new message for the user arrives its send to the broker in form of a json-object. How ever, when using the normale syntax I won't get any results (since the code waits for the connection to be closed as I suppose):

return this._http.request(req).map( (res: Response) => {    
    return res.json();
}).subscrube( results => {
    console.log("results arrived", results);
});

Consuming streams has a lot of advantages (like video streaming with a rest api, awesome). Is there a way to do this?

/edit: A little bit more information here: I checked the http transaction and they look okay:

GET /broker HTTP/1.1
User-Agent: curl/7.35.0
Host: mydomain.com:12345
Accept: */*
Authorization: Bearer [JWT]

The server ansers with the response header:

HTTP/1.1 200 OK
Content-Type: application/json; charset=utf-8
Date: Mon, 01 Feb 2016 11:16:45 GMT
Transfer-Encoding: chunked

[then the client is listening on this connection until the server pushes something into the stream

ba
{"target":"56addcef7bec6d5785ee6945","payload": ...}

Seems correct so far. I can see the size of the chunk in hex (ba), the CRLF, the payload (json) and the final CRLF. So, when talking about HTTP Standard everything seems okay. However, my Angular2 won't fetch the chunks.

Answer 1

As already mentioned in one of the comments, this is currently not possible with Angular2 itself, a feature request is already documented (from 06/2015): check it out here

To get things working until the feature is implemented I now Oboe.js as an external API.

Add dependency in package.json:

"dependencies": {
    "oboe": "*",
     ...
},

Don't forget to run npm install or to download the package etc.

Include oboe in index.html

<script src="node_modules/oboe/dist/oboe-browser.min.js"></script>

Use oboe to consume the http-stream

import {Injectable, OnInit} from 'angular2/core';

// Declare oboe as a Variable
declare var oboe: any;

@Injectable()
export class BrokerService {
    private oboeService: any;
    constructor() {
        this.listen();
    }

    listen(): void {
        console.log("registering message broker")
        var config = {
            'url': 'https://you-streaming-api.com:123/broker',
            'method': "GET",          
            'headers': {'Authorization': 'Bearer ' + this._jwToken },         
            'body': '',     
            'cached': false,
            'withCredentials': true            
        }            
        this.oboeService = oboe(config);            
        // The '!' will only consume complete json objects
        this.oboeService.node ('!', function (thing) {            
            console.log("new broker message", thing); 
    });
    }
}

Answer 2

I think you need to implement a custom backend that forwards each event it receives individually instead of only the whole result.

This alternative backend can be registerd for DI to be used everywhere instead of the original one, or only when explicitely requested. The later would also need an alternative Http implementation to distinguish between default and streaming one because this type is the one actually injected into a component or service. The backend is only a transitive dependency requested by Http.

Answer 3

In my case const oboe = require('oboe'); worked instead of declare var oboe: any;

Env: Ionic 5 with Angular 9

Updated code:

const oboe = require('oboe');   <<<---- Updated code

@Injectable()
export class BrokerService {
    private oboeService: any;
    constructor() {
        this.listen();
    }

    listen(): void {
        console.log("registering message broker")
        var config = {
            'url': 'https://you-streaming-api.com:123/broker',
            'method': "GET",          
            'headers': {'Authorization': 'Bearer ' + this._jwToken },         
            'body': '',     
            'cached': false,
            'withCredentials': true            
        }            
        this.oboeService = oboe(config);            
        // The '!' will only consume complete json objects
        this.oboeService.node ('!', function (thing) {            
            console.log("new broker message", thing); 
    });
    }
}