How to restart or refresh an Observable?

Question

I've got a TypeScript/Angular 2 Observable that works perfectly the first time I call it. However, I'm interested in attaching multiple subscribers to the same observable and somehow refreshing the observable and the attached subscribers. Here's what I've got:

query(): Rx.Observable<any> {

return this.server.get('http://localhost/rawData.json').toRx().concatMap(
    result =>
        result.json().posts
    )
    .map((post: any) => {
        var refinedPost = new RefinedPost();
        refinedPost.Message = post.Message.toLowerCase();

        return refinedPost;

    }).toArray();

}

Picture that there is a refresh button that when pressed, re-executes this observable and any subscribers that are connected to it get an updated set of data.

How can I accomplish this?

Answer 1

I don't know so much about Angular2 and Typescript, but typescript being a superset of javascript, I made the following hypothesis (let me know if I am wrong) which should make the following javascript code work :

• server.get returns a promise (or array or Rx.Observable)
• posts is an array of post

You would need to create an observable from the click event on that button, map that click to your GET request and the rest should be more or less as you wrote. I cannot test it, but it should go along those lines:

// get the DOM id for the button
var button = document.getElementById('#xxx');

// create the observable of refined posts
var query$ = 
    Rx.Observable.fromEvent(button, 'click')
        .flapMapLatest(function (ev) {
                     return this.server.get('http://localhost/rawData.json')
                   })
        .map(function (result){return result.json().posts})
        .map(function (posts) {
                return posts.map(function(post){
                                   var refinedPost = new RefinedPost();
                                   refinedPost.Message = post.Message.toLowerCase();
                                   return refinedPost;
                           })
             })
        .share()

// subscribe to the query$. Its output is an array of refinedPost
// All subscriptions will see the same data (hot source)
var subscriber = query$.subscribe(function(refinedPosts){console.log(refinedPosts)})

Some explanation :

1. Every click will produce a call to server.get which returns an observable-compatible type (array, promise, or observable). That returned observable is flattened to extract result from that call
2. Because the user can click many times, and each click generate its flow of data, and we are interested (another hypothesis I make here too) only in the result of the latest click, we use the operator flatMapLatest, which will perform the flatMap only on the observable generated by the latest click
3. We extract the array of post posts and make an array of refinedPost from it. In the end, every click will produce an array of refinedPost which I assume is what you want
4. We share this observable as you mention you will have several subscribers, and you want all subscribers to see the same data.

Let me know if my hypotheses are correct and if this worked for you.

In addition I recommend you to have a look at https://gist.github.com/staltz/868e7e9bc2a7b8c1f754

In addition to being a very good reminder of the concepts, it addresses a refresh server call problem very similar to yours.