Is this a reasonable implementation of a cancelable Observable?

Question

I have a multi-step process - a callback-taking function, and two promise-returning functions - and I'd like to construct an Observable around it that supports being canceled. What I've got is:

let observableFromEvent = evt => Observable.create(({ next, error, complete }) => {
    let canceled
    Promisify(callbackFn)(evt)
       .then(r => !canceled && promiseF1(r))
       .then(r => !canceled && promiseF2(r))
       .then(next)
       .then(complete, error)
    return () => {
        canceled = true
    }
})

Can anyone tell me if I've left anything out? Or left too much in? My intended usage is:

event$.switchMap(observableFromEvent)

to create a series of Observables which cancel their previous when they are spawned.

Answer 1

You can do it with builtin operators, as simple as following:

observableFromEvent = evt => Observable
  .bindCallback(callbackFn)(evt)
  .flatMap(promiseF1)
  .flatMap(promiseF2)