How to mimic a "callback with promise" function parameter in rxjs?

Question

I'm an absolute rxjs beginner. For me to start learning to think in observables, I need to translate concepts through code examples. I feel like if I can see the code for this, I can start to do this on my own with other concepts.

I do NOT want to CONVERT a promise to an observable, I want to make a new implementation using Observable that can behave like a promise. How would I re-write the following using Observables?

 constructor(){
    let makeMessage2 = function(){
        return new Promise(resolve, reject){
           setTimeout(()=>{
                  var r = Math.random();
                  resolve("message two plus random value: " + r );
           }, 1000);
        }
    }
    this.logMessageAndResultOfCallback("message one!", makeMessage2);
}
private sideEffect1:string = "";
private sideEffect2:string = "";

logMessageAndResultOfCallback( message1:string, callback:Function ){
    console.log(message1);
    this.sideEffect1 = message1;

    callback().then((message2)=>{
          console.log(message2);
          this.sideEffect2 = message2;
    }
}

I guess the part I don't understand is how to define the "callback" function, how to invoke it. I understand I would wait for the complete or emit handlers, like makeMessage2().subscribe(message2 => console.log(message2)); but I don't have any clue how to define makeMessage2.

This may be totally clueless question, but I've read about 10 different intros to rxjs and this hasn't quite clicked. I just need to map this scenario to observable pattern and I think I can understand it.

Basically, I want to define an observable function myObs() that does not "execute immediately" but "executes" whenever the someMethod(message:string,obs:Observable) is executed. When myObs is executed, it should do something ansynchronously within it (like get the result of a HTTP request) then set the next value, then fire a complete() so my observer defined in someMethod can handle the complete and do something with the result.

Edit: I'm not concerned with the timer or native equivalents in rxjs, this is just to simulate any async action, like getting data from the server.

Answer 1

The code that you wrote and want to 'translate' to observables probably would not work. callback is a promise, not a function, so you can't write callback().

Did you try this introduction too? It worked for many people.

To answer your question, you could write

Rx.Observable.of(""message one!", "message two!")
  .map(console.log.bind(console)) // would be better to use `do` operator actually, for semantic reasons, but that works the same here
  .subscribe(noop, noop, noop)

or

Rx.Observable.of(""message one!", "message two!")
  .subscribe(console.log.bind(console), noop, noop)

where noop is a function that does not do anything, i.e. function noop(){}

In short, your stream emits data, that data flow through a series of operators, and the data flow is started by .subscribe. In your case, you have nothing interesting to do on subscription, because all you do is logging.

Rxjs Streams are in fact callback-based under the hood. You want to check this answer to understand it.

Answer 2

I solved it with help from this guide.

import {Observable} from 'rxjs';
var makeMessage2 = Observable.create(observer => {
  // Yield a single value and complete
  setTimeout(function(){
     let r = Math.random();
     observer.next("message two plus random value: " + r );
     observer.complete();
  }, 1000);
  return () => console.log('disposed')
});
logMessageAndResultOfCallback( "some message one", makeMessage2);


logMessageAndResultOfCallback( message1:string, callback:Observeable ){
    console.log(message1);
    this.sideEffect1 = message1;

    var subscription = callback.subscribe(
       (value)=>{this.sideEffect2 = value;},
       (e) =>{ console.log('onError: %s', e)},
       () => {console.log(this.sideEffect2);});
      subscription.dispose();
}