title: In RXJS, how do I force re-execution of an Observable created with bindNodeCallback using retryWhen?
body: <p>I have a <code>Subject</code> that receives objects and then makes a call to a function that I have wrapped with <code>bindNodeCallback</code>. I'd like to retry execution of that function, using <code>retryWhen</code>, if it fails due to an error or if the result comes back false.</p>
  <p>I've tried a few different approaches but haven't been successful at getting the bound function to fire off again.</p>
  <p>I have a <a href="https://codesandbox.io/s/youthful-wave-u5bhx?file=/src/index.ts" rel="nofollow noreferrer">codesandbox setup here</a>.</p>
  <pre class="lang-js prettyprint-override"><code>function fakeSend(
    task: string,
    cb: (err: Error | null, result?: boolean) =&gt; void
  ) {
    console.log(&quot;fakesend&quot;, task);
  
    setTimeout(() =&gt; {
      const hasError = Math.random() &lt; 0.5;
      const res = Math.random() &lt; 0.5;
      console.log(hasError ? &quot;hasError&quot; : `responding with ${res}`);
      if (hasError) {
        return cb(new Error(&quot;error&quot;));
      }
      return cb(null, res);
    }, 100);
  }
  
  const boundSend = bindNodeCallback(fakeSend);
  const subject = new Subject&lt;string&gt;();
  
  subject.subscribe(
    (task) =&gt; {
      boundSend(task)
        .pipe(
          tap((status) =&gt; {
            if (!status) {
              throw new Error(&quot;Did not send&quot;);
            }
            return status;
          }),
          retryWhen((errs) =&gt;
            errs.pipe(
              delay(1000),
              tap((err) =&gt; console.log)
            )
          )
        )
        .subscribe({
          next: console.log,
          error: console.error,
          complete: () =&gt; {
            console.log(&quot;complete&quot;, task);
          }
        });
    },
    (error) =&gt; {
      console.log(&quot;error in subject subscribe&quot;);
    }
  );
  
  subject.next(&quot;test1&quot;);
  subject.next(&quot;test2&quot;);
  subject.next(&quot;test3&quot;);
  subject.next(&quot;test4&quot;);
  
    
  </code></pre>
  <p>Thanks!</p>
  <p><div class="snippet" data-lang="js" data-hide="false" data-console="true" data-babel="false">
  <div class="snippet-code">
  <pre class="snippet-code-js lang-js prettyprint-override"><code>const { Subject, bindNodeCallback } = rxjs;
  const { tap, retryWhen, delay } = rxjs.operators;
  
  function fakeSend(
    task,
    cb
  ) {
    console.log("fakesend", task);
  
    setTimeout(() =&gt; {
      const hasError = Math.random() &lt; 0.5;
      const res = Math.random() &lt; 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();
  
  subject.subscribe(
    (task) =&gt; {
      boundSend(task)
        .pipe(
          tap((status) =&gt; {
            console.log('tap');
            if (!status) {
              throw new Error("Did not send");
            }
            return status;
          }),
          retryWhen((errs) =&gt;
            errs.pipe(
              delay(1000),
              tap((err) =&gt; console.log)
            )
          )
        )
        .subscribe({
          next: console.log,
          error: console.error,
          complete: () =&gt; {
            console.log("complete", task);
          }
        });
    },
    (error) =&gt; {
      console.log("error in subject subscribe");
    }
  );
  
  subject.next("test1");
  subject.next("test2");
  subject.next("test3");
  subject.next("test4");</code></pre>
  <pre class="snippet-code-html lang-html prettyprint-override"><code>&lt;script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/6.6.2/rxjs.umd.min.js"&gt;&lt;/script&gt;</code></pre>
  </div>
  </div>
  </p>
  <h3>edit</h3>
  <p>I got it working with this:</p>
  <pre class="lang-js prettyprint-override"><code>subject.subscribe(
    (task) =&gt; {
      of(task).pipe(
        mergeMap((v) =&gt; boundSend(v)),
        tap((val) =&gt; {
          if (!val) throw new Error(&quot;Did not send&quot;);
        }),
        retryWhen((errs) =&gt; errs.pipe(delay(300)))
      ).subscribe();
    });
  </code></pre>
  <p>However, I don't understand why this doesn't work:</p>
  <pre class="lang-js prettyprint-override"><code>subject
    .pipe(
      mergeMap((task) =&gt; boundSend(task)),
      tap((val) =&gt; {
        if (!val) throw new Error(&quot;Did not send&quot;);
      }),
      retryWhen((errs) =&gt; errs.pipe(delay(300)))
    )
    .subscribe();
  </code></pre>
  <h2>Edit2</h2>
  <p>Well. I guess this explains it:  <a href="https://github.com/ReactiveX/rxjs/issues/1401" rel="nofollow noreferrer">https://github.com/ReactiveX/rxjs/issues/1401</a></p>
  <p>Here I was thinking it was because I'm an RX noob :/</p>