Basic question on rxjs observables --how to feed data to an observable 'by hand'?

Question

I would like to use an observable to signal various parts of my Angular application about "exceptional states" and likewise, but I realize I don't really understand how they work.

In the following code, I built an observer object, and created an observable from it. What I would like to do is figure out how to call the "next" method outside the scope of the Observable.create method so I can insert arbitrary events into the stream. Calling next directly on the observer deosn't seem to be the answer.

 var observer = {
      next: function(value) {
        this.myvalue="last value: " + value;
      },
      error: function(error) {
        console.log(error);
      },
      complete: function() {
        console.log('completed');
      },
      myfunction: function() {
        this.myvalue = "Penguins"
      },
      myvalue: ""
    }

   let myobs$ : Observable<any> = Observable.create(function(obs) {
      obs.next("Some foo");

      obs.next("Other foo");
    })

    let foo=myobs$.subscribe((observer)=> {
      console.log("This is the subscription: ", observer)
    })

    setTimeout(function() {
      observer.next("This is a late breaking value");
      console.log(observer.myvalue);
    }, 2000);

  }

this code produces the following console output:

This is the subscription:  Some foo
This is the subscription:  Other foo
last value: This is a late breaking value

So it looks like calling next on the observer object directly (which I tried inside of the timeout function at the bottom) doesn't produce a value inside the subscription.

Also clear I guess that I don't understand how these things are supposed to work. It would be helpful to understand that if I set up an observerable and I want to 'manually' insert data into the stream to be picked up by subscribers, how exactly do I do that? I can see how you do it with event-propagating things like mouse clicks or ajax requests, but what I'm looking to do is create a stream that I can feed on an ad-hoc basis when certain interesting things happen in various places in my code.

Answer 1

You should read on RxJs Subject. That will be what you need. From the documentation:

A Subject is a special type of Observable that allows values to be multicasted to many Observers. Subjects are like EventEmitters.

There are different types of subjects. ReplaySubject, BehaviorSubject, AsyncSubject. I suggest you dive into the documentation or some tutorials on how and when to use them. From your observable you also want the previous values and new value to be emitted. You can make such custom behavior with the pipable operators that RxJs ships with. I suggest you also read on those. Here is another post on stackoverflow that exactly does what you want:

Based on that example:

const subject = new Subject().pipe(
  startWith('Penguins'), // emitting first value to fill-in the buffer
  pairwise(),
  map([previous, current] => {
    return previous + current;
  }),
);

const observer = {
  next: (value) => {
    console.log(value);
  },
  error: (error) => {
    console.log(error);
  },
  complete: () => {
    console.log('completed');
  },
};

// Subjects
const subscriber = subject.subscribe(observer);

subject.next('Some foo');