Creating and returning Observable from Angular 2 Service

Question

UPDATE: 9/24/16 Angular 2.0 Stable

This question gets a lot of traffic still, so, I wanted to update it. With the insanity of changes from Alpha, Beta, and 7 RC candidates, I stopped updating my SO answers until they went stable.

This is the perfect case for using Subjects and ReplaySubjects

I personally prefer to use ReplaySubject(1) as it allows the last stored value to be passed when new subscribers attach even when late:

let project = new ReplaySubject(1);

//subscribe
project.subscribe(result => console.log('Subscription Streaming:', result));

http.get('path/to/whatever/projects/1234').subscribe(result => {
    //push onto subject
    project.next(result));

    //add delayed subscription AFTER loaded
    setTimeout(()=> project.subscribe(result => console.log('Delayed Stream:', result)), 3000);
});

//Output
//Subscription Streaming: 1234
//*After load and delay*
//Delayed Stream: 1234

So even if I attach late or need to load later I can always get the latest call and not worry about missing the callback.

This also lets you use the same stream to push down onto:

project.next(5678);
//output
//Subscription Streaming: 5678

But what if you are 100% sure, that you only need to do the call once? Leaving open subjects and observables isn’t good but there’s always that “What If?”

That’s where AsyncSubject comes in.

let project = new AsyncSubject();

//subscribe
project.subscribe(result => console.log('Subscription Streaming:', result),
                  err => console.log(err),
                  () => console.log('Completed'));

http.get('path/to/whatever/projects/1234').subscribe(result => {
    //push onto subject and complete
    project.next(result));
    project.complete();

    //add a subscription even though completed
    setTimeout(() => project.subscribe(project => console.log('Delayed Sub:', project)), 2000);
});

//Output
//Subscription Streaming: 1234
//Completed
//*After delay and completed*
//Delayed Sub: 1234

Awesome! Even though we closed the subject it still replied with the last thing it loaded.

Another thing is how we subscribed to that http call and handled the response. Map is great to process the response.

public call = http.get(whatever).map(res => res.json())

But what if we needed to nest those calls? Yes you could use subjects with a special function:

getThing() {
    resultSubject = new ReplaySubject(1);

    http.get('path').subscribe(result1 => {
        http.get('other/path/' + result1).get.subscribe(response2 => {
            http.get('another/' + response2).subscribe(res3 => resultSubject.next(res3))
        })
    })
    return resultSubject;
}
var myThing = getThing();

But that’s a lot and means you need a function to do it. Enter FlatMap:

var myThing = http.get('path').flatMap(result1 => 
                    http.get('other/' + result1).flatMap(response2 => 
                        http.get('another/' + response2)));

Sweet, the var is an observable that gets the data from the final http call.

OK thats great but I want an angular2 service!

I got you:

import { Injectable } from '@angular/core';
import { Http, Response } from '@angular/http';
import { ReplaySubject } from 'rxjs';

@Injectable()
export class ProjectService {

  public activeProject:ReplaySubject<any> = new ReplaySubject(1);

  constructor(private http: Http) {}

  //load the project
  public load(projectId) {
    console.log('Loading Project:' + projectId, Date.now());
    this.http.get('/projects/' + projectId).subscribe(res => this.activeProject.next(res));
    return this.activeProject;
  }

 }

 //component

@Component({
    selector: 'nav',
    template: `<div>{{project?.name}}<a (click)="load('1234')">Load 1234</a></div>`
})
 export class navComponent implements OnInit {
    public project:any;

    constructor(private projectService:ProjectService) {}

    ngOnInit() {
        this.projectService.activeProject.subscribe(active => this.project = active);
    }

    public load(projectId:string) {
        this.projectService.load(projectId);
    }

 }

I’m a big fan of observers and observables so I hope this update helps!

Original Answer

I think this is a use case of using a Observable Subject or in Angular2 the EventEmitter.

In your service you create a EventEmitter that allows you to push values onto it. In Alpha 45 you have to convert it with toRx(), but I know they were working to get rid of that, so in Alpha 46 you may be able to simply return the EvenEmitter.

class EventService {
  _emitter: EventEmitter = new EventEmitter();
  rxEmitter: any;
  constructor() {
    this.rxEmitter = this._emitter.toRx();
  }
  doSomething(data){
    this.rxEmitter.next(data);
  }
}

This way has the single EventEmitter that your different service functions can now push onto.

If you wanted to return an observable directly from a call you could do something like this:

myHttpCall(path) {
    return Observable.create(observer => {
        http.get(path).map(res => res.json()).subscribe((result) => {
            //do something with result. 
            var newResultArray = mySpecialArrayFunction(result);
            observer.next(newResultArray);
            //call complete if you want to close this stream (like a promise)
            observer.complete();
        });
    });
}

That would allow you do this in the component:
peopleService.myHttpCall('path').subscribe(people => this.people = people);

And mess with the results from the call in your service.

I like creating the EventEmitter stream on its own in case I need to get access to it from other components, but I could see both ways working…

Here’s a plunker that shows a basic service with an event emitter: Plunkr

Leave a Comment Cancel reply