How to subscribe to stream in Angular2?

Question

I'm having my streaming web-service running on localhost:8080/stream, which response when any new message added to subscribed mqtt stream. I want to consume this web-service in my Angular2 app. I'm using RxJS to consume NodeJS APIs in Angular2. I tried following code which calls localhost:8080/stream once and ends response. I want my observable to listen continuously to web-service.

var headers = new Headers();
headers.append('Access-Control-Allow-Origin', '*');
headers.append('Content-Type', 'application/json');

let options = new RequestOptions({ headers: headers }); // Create a request option
 return this.http.get("http://localhost:8080/stream", options) // ...using post request
   .map((res: Response) => res.json()) // ...and calling .json() on the response to return data
   .catch((error: any) => Observable.throw(error.json().error));

Possible duplicate of [RxJs Observable with WebSocket](https://stackoverflow.com/questions/43328451/rxjs-observable-with-websocket) — angularrocks.com, Aug 30 '17 at 05:37

Dheeraj Kumar · Answer 1 · 2017-08-30T05:24:13.543

1

If I understand your question right, you want to consume data from stream where new messages arrive at some period of time.

To achieve this You need add subscribe to the service.

return this.http.get("http://localhost:8080/stream", options) // ...using post request
   .map((res: Response) => res.json()) // ...and calling .json() on the response to return data
   .catch((error: any) => Observable.throw(error.json().error)
   .subscribe(result => this.result =result));

Result will be updated as new data arrives, and you can use it the way want.

Note: It is best practice to make http calls separate in services and subscribe the service in your component.

For your reference I am adding an example I have worked on for demo purpose.

Create a service for http calls

@Injectable() export class JsonPlaceHolderService{

    constructor(private http:Http){}

    getAllPosts():Observable<Post[]>{

        return this.http.get("https://jsonplaceholder.typicode.com/posts")
        .map(res=>res.json())


    }

}

From your component call service and listen to changes continuously.

export class PostsComponent implements OnInit{

constructor(private _service:JsonPlaceHolderService){}

jphPosts:Post[];

title:string="JsonPlaceHolder's Post data";

ngOnInit():void{

    this._service.getAllPosts().subscribe(
        (data) =>this.jphPosts = data,
        (err) => console.log(err),
        ()=>console.log("service call completed")
    );

}

}

edited Aug 30 '17 at 05:24

answered Aug 30 '17 at 05:08

Dheeraj Kumar

3,917
8
43
80

@VolodymyrBilyachat If I have missed something, you are welcome to advise :) – Dheeraj Kumar Aug 30 '17 at 05:11
"I want my observable to listen continuously to web-service." So OP wants to subscribe to long pooling or websocket – Vova Bilyachat Aug 30 '17 at 05:12
it's not running over websocket, it's simply running on http server. – netfreak30 Aug 30 '17 at 05:14
Isn't thats what Subscribing with Observable does? Listen continuously to api or service? – Dheeraj Kumar Aug 30 '17 at 05:14
I'm new to rxjs and angular2, so, I don't have clear idea about it. but my code is not doing so. – netfreak30 Aug 30 '17 at 05:15
correct me if im wrong as far as i understand get or post only listens for a single response – shaN Aug 30 '17 at 05:16
1

@shaN When you use promise it gets you one response, but when using Observables and subscribe, you get response in stream. – Dheeraj Kumar Aug 30 '17 at 05:20
@DarshanaAhalpara I have added an example for you. – Dheeraj Kumar Aug 30 '17 at 05:20
Yes, I'm trying it. – netfreak30 Aug 30 '17 at 05:23
@DarshanaAhalpara Are you getting any error? Have you received data from api? – Dheeraj Kumar Aug 30 '17 at 05:45
I'm not getting any error, getting response code 200. – netfreak30 Aug 30 '17 at 05:46
But I don't want response to end till I unsubscribe. – netfreak30 Aug 30 '17 at 05:47
It wont unsubcribe., but it will keep listening to changes. – Dheeraj Kumar Aug 30 '17 at 05:52
I'm getting this on my browser console : zone.js:2019 XHR failed loading: OPTIONS "http://localhost:8080/stream". – netfreak30 Aug 30 '17 at 06:25
Add "Access-Control-Allow-Origin: * Access-Control-Allow-Methods: POST, GET, OPTIONS" to http header – Dheeraj Kumar Aug 30 '17 at 06:28

shaN · Answer 2 · 2017-08-30T06:48:52.620

Streaming data from nodejs to angular with socket.io

This is something that would have been of great use when I was trying to do this. Following contains code from a socket.io package for angular credit goes to original author. This is taken from a working solution and may need some tweaking.

Server side

var app     = express(),
    http    = require('http'),
    ioServer    = require('socket.io');

var httpServer = http.createServer(app);
var io = new ioServer();
httpServer.listen(1337, function(){
  console.log('httpServer listening on port 1337');
});

io.attach(httpServer);

io.on('connection', function (socket){
    console.log(Connected socket ' + socket.id);
});


//MQTT subscription
client.on('connect', function () { 
    client.subscribe(topic, function () { 
        console.log("subscribed to " + topic) 
        client.on('message', function (topic, msg, pkt) { 

            io.sockets.emit("message", {topic: topic, msg: msg, pkt: pkt});

        }); 
    }); 
});

Client Side

Create a customService in angular with following

import * as io from 'socket.io-client';

declare

private ioSocket: any;
private subscribersCounter = 0;

inside service class

constructor() {

    this.ioSocket = io('socketUrl', {});

}

on(eventName: string, callback: Function) {
    this.ioSocket.on(eventName, callback);
}

removeListener(eventName: string, callback?: Function) {
    return this.ioSocket.removeListener.apply(this.ioSocket, arguments);
}

fromEvent<T>(eventName: string): Observable<T> {
    this.subscribersCounter++;
    return Observable.create((observer: any) => {
        this.ioSocket.on(eventName, (data: T) => {
            observer.next(data);
        });
        return () => {
            if (this.subscribersCounter === 1) {
                this.ioSocket.removeListener(eventName);
            }
        };
    }).share();
}

In your component, Import customService as service

service.on("message", dat => {
 console.log(dat);
});

this.client = mqtt.connect() // you add a ws:// url here, -- but I'm having http url only. — netfreak30, Aug 30 '17 at 05:51
yes you have to import that on top `import { * } from 'mqtt';` — shaN, Aug 30 '17 at 06:11
url will be of mqtt broker? and subscribe topic of mqtt in angular2 itself? — netfreak30, Aug 30 '17 at 06:13
Let us [continue this discussion in chat](http://chat.stackoverflow.com/rooms/153198/discussion-between-shan-and-darshana-ahalpara). — shaN, Aug 30 '17 at 06:13

score 0 · Answer 3 · answered Aug 30 '17 at 06:27

0

You should use websocket on angular and make it listen to your service URL after that you should listen to its events (open, close, message) and then create your own Subject stream using Rxjs to push the new data to the subscribers.

Please check the URL below:

https://medium.com/@lwojciechowski/websockets-with-angular2-and-rxjs-8b6c5be02fac

answered Aug 30 '17 at 06:27

Nour

5,609
3
21
24

I already tried that, but it's not working for me as it requires ws url and I'm having http url. – netfreak30 Aug 30 '17 at 06:41

How to subscribe to stream in Angular2?

3 Answers3