Can't repeat or retry Observable.bindCallback and Observable.bindNodeCallback

[trxcllnt]

Observable.bindCallback and Observable.bindNodeCallback create internal AsyncSubjects that live forever after the first subscription, which means we can't retry or repeat these Observables.

I understand this is how it worked before, but that behavior seems to be incidentally caused by the old multicast behavior disallowing retrying and repeating. Does anybody else think we should allow retrying/repeating, or at least allow end users to override the multicast behavior?

import Rx from 'rxjs';

const getObs = Rx.Observable.bindCallback(
  (msg, cb) => {
    console.log('trying callback');
    cb(new Error(msg));
  },
  (error) => { throw error; }
);

// only prints `trying callback` once when expected to print 3 times

getObs('some error')
  .retry(3)
  .subscribe(null, (finalError) => {
    console.log("got error:", finalError);
  });

[benlesh]

I see what you mean. It's probably a name thing. Should it have been named publishCallback or publishedCallback? @staltz? @kwonoj? @mattpodwysocki?

... in a few weeks, we're locking these names in stone until another major, which hopefully won't happen for a while.

[mattkime]

I'd really like to see this fixed.

[staltz]

Does anybody else think we should allow retrying/repeating, or at least allow end users to override the multicast behavior?

Not me. AFAIK go ahead.

[mattkime]

ignore me. my knowledge of rxjs is rapidly increasing.

[vvenegasv]

I am new on this, but i think i have the same issue. This code is called only once:

let observableMapMove = Rx.Observable.bindCallback(<(a: number) => any>this.map.on);
        let result = observableMapMove((<any>window).plugin.google.maps.event.CAMERA_CHANGE);
        result.subscribe(x => console.log('Hello world --> ' + JSON.stringify(x)));

But this code is called many many times

this.map.on((<any>window).plugin.google.maps.event.CAMERA_CHANGE, () => {
          try {
            console.log('hello world!');
          } catch(err) {
            console.log(err);
          }          
        });

How can i fixed ?

[kwonoj]

@vvenegasv This issue is a discussion about repeat-retry ability on bindNodecallback, and code snippet you suggested is not related to the current case. I can't say for sure by given code snippet but map.on looks more of eventListener type.

[vvenegasv]

@kwonoj Ok, you are right, this is an event.

I did it like this, because with "fromEvent" don't work at all. I was thinking that is because it's not a standard event.

Excuse me if my question is silly, but i want to know if i attach an event with bindCallback method, it will be raised everytime? or only once?

PS: Sorry for my bad english, i wrote the best i can.

[kwonoj]

Excuse me if my question is silly,

no worries, it is no silly at all 👍

• for event types, you could use fromEvent, or fromEventPattern to compose custom type of eventlistener.
• bindcallback will be invoked once. Think of typical callback based async, i.e fs.readFile.

As I think your case is not tied to this issue, I'd like to suggest to either create new issue if you think current RxJS behavior is problematic, or use SO (http://stackoverflow.com/questions/tagged/rxjs5) for general questions.

[vvenegasv]

@kwonoj Yes you are right. Thanks a lot! :)

[mattflix]

I too find it very odd that calling subscribe() on the Observable created by bindCallback() does not re-trigger the callback. This behavior completely breaks otherwise normal repeat/retry logic in the midst of a larger Observable sequence. I always have to remember to wrap the bindCallback() in defer() or something similar to get it to trigger again -- which seems unintuitive and is often forgotten.

Just like with any "normal" Observable, the default expectation should be that calling subscribe() will re-do the work. IMO, that would be the Least Surprising thing to happen.

If one desires to share/cache work, then publish() or some other technique can be used (which would also then serve to explicitly document the intended behavior).

[benlesh]

@mattflix, @trxcllnt el al: There's a couple things to discuss here:

1. a bindCallback observable can't be retried/replayed.
2. a bindCallback observable only calls the callback once and caches the result.

I think 1 is an issue. We should be able to retry or replay them, or we should rename this to publishCallback, since publish variants can't be retried or replayed.

2, on the other hand is stickier. This method was created to model Rx 4's fromCallback... since the behavior was different from other fromX methods in that it made a hot observable that cached the result, we renamed it to bindCallback. I don't think I want to change the behavior of this, BUT, I'm all for adding a fromCallback method that creates a cold observable, from which we can compose any behavior we want.

Thoughts?

[mattflix]

I am also for bringing back fromCallback in a form that works "correctly" -- as a simple, cold observable that does exactly what re-invoking the underlying callback API would do (i.e., does the work over, every time).

I would recommend renaming bindCallback to publishCallback to reduce confusion and set expectations about what it actually does. That said, there doesn't seem to be any "need" for a publishCallback (other than as a convenience) since the same behavior could always be composed.

If it were me, I would just nix bind/publishCallback and just have a nice, cold fromCallback.

---

On related question... why was the context argument removed? The Rx4 version of fromCallback took an optional context parameter to be passed as the this pointer of the invoked callback API. With the current Rx5, I must first manually api.method.bind(api) instance method callback APIs before passing them to bindCallback. That itself (the two uses of the word "bind", plus the need for both "bind" operations) could also be a source of unnecessary confusion.

[larssn]

Thought I would chime in with my thoughts (and reawaken this thread). I'm surprised that retry doesn't work on bound callbacks. I'm wrapping an old JS library in Rx, with bindNodeCallback, and it seems weird to me that if the callback fails, It'll just repeat the failure x amount of times.

Gonna have to use something other than bindNodeCallback for now, I suppose.

[trxcllnt]

@larssn wrapping in defer is a reasonable workaround, but it is annoying.

[benlesh]

@trxcllnt ... well if we want to "fix" this, we could propose a new method. The one that exists is largely to honor legacy behavior from RxJS 4. 🤷‍♂️

[benlesh]

IIRC, we even named it "bind" to make it sound more like it was locked in to one result and done.

[larssn]

@trxcllnt Thanks, I ended up cannibalising the source, and stripped out the subject.

Maybe call it fromNodeCallback again, I can see you're closing in on version 6 anyway.

[larssn]

Is it possible to revisit this discussion, now that we're in RxJS6?

[jsgoupil]

I see a lot of people are proposing to use defer() and most likely repeat(). Because there is not a lot of code examples here, I spent quite a bit of time reading all these comments. Hopefully I can help people here:

// Don't blindly copy this, read further
defer(() =>
    bindCallback(myObject.on)("click")
)
.pipe(repeat())
.pipe(takeUntil(this.destroySubject))
.subscribe(evt => { });

The problem with this, is that you attach plenty of on events, without calling off. The garbage collector might take care of it at some point, but if you keep your page running for a while, you might start hogging some memory.

You should use bindCallback if it's a one event thing. Something like .one, which triggers only ONCE.

Now, if you are about to use a on/off pattern, or the event type, you should rather use a fromEvent/fromEventPattern!

fromEvent<IEvent>(myObject, 'click')
    .pipe(takeUntil(this.destroySubject))
    .subscribe(evt => { });

fromEventPattern<IEvent>(handler => {
    myObject.on('click', handler);
}, handler => {
    myObject.off('click, handler);
})
    .pipe(takeUntil(this.destroySubject))
    .subscribe(evt => { });

I think there is just a misconception about this bindCallback, maybe the documentation should be more explicit. People come to this thread because they see a problem, but they are not on the right track. Now, it could also be a naming issue, not saying this bug is not valid.

[trxcllnt]

@jsgoupil yes, bindCallback is intended to help translate node-style callback APIs into Observables and fromEvent is intended to translate event emitters/dispatchers/delegates into Observables. I'm not sure what the docs say on this topic, but that should be noted explicitly if it's not.

[chanced]

What is the recommended solution for this?

I've been battling with it all day, thinking it was because I still have no idea wtf I'm doing with RX 🥴. I have a codesandbox where I've been trying to figure this out here (link)

function fakeSend(
  task: string,
  cb: (err: Error | null, result?: boolean) => void
) {
  console.log("fakesend", task);
  setTimeout(() => {
    const hasError = Math.random() < 0.5;
    const res = Math.random() < 0.5;
    console.log(hasError ? "hasError" : `responding with ${res}`);
    if (hasError) {
      return cb(new Error("error"));
    }
    return cb(null, res);
  }, 100);
}
const boundSend = bindNodeCallback(fakeSend);

const subject = new Subject<string>();
subject
  .pipe(
    mergeMap((v) => boundSend(v)),
    tap((res) => {
      console.log("res", res);
      if (!res) throw new Error("did not send");
    }),
    retryWhen((e) => e.pipe(delay(100)))
  )
  .subscribe();

subject.next("test1");
subject.next("test2");
subject.next("test3");
subject.next("test4");

The only way I've been able to get it to refire is by doing this:

subject.subscribe((task) => {
  of(task).pipe(
    mergeMap((v) => boundSend(v)),
    tap((val) => {
      if (!val) throw new Error("Did not send");
    }),
    retryWhen((errs) => errs.pipe(delay(300)))
  ).subscribe();
});

But that doesn't seem right at all.

Thank you :)

[benlesh]

@chanced you could bind it inside your mergeMap.

[chanced]

@benlesh Not sure what I'm missing, is this right?

const subject = new Subject<string>();
subject
  .pipe(
    mergeMap(res => bindNodeCallback(fakeSend)(res)),
    tap(res => {
      if (!res) throw new Error("did not send");
    }),
    retryWhen(e => e.pipe(delay(100)))
  )
  .subscribe();

Because that's not re-executing for me.

https://codesandbox.io/s/youthful-wave-u5bhx?file=/src/index.ts

[chanced]

Before I found this thread, I had posted over on stack. I recently got a really great answer. incase anyone runs into this issue in the future.

also @benlesh as an aside, I really hope I didn't come across as snarky. I just reread what I wrote and went "aarg that sounded rude" but that wasn't my intention at all. I apologize if it came across that way.