tag:blogger.com,1999:blog-6753769565491687768.post132278296001058002..comments2019-02-21T06:29:44.506+01:00Comments on Tomasz Nurkiewicz around Java and concurrency: Detecting and testing stalled streams - RxJava FAQTomasz Nurkiewicznoreply@blogger.comBlogger2125tag:blogger.com,1999:blog-6753769565491687768.post-50811124295087818482017-09-15T16:18:05.476+02:002017-09-15T16:18:05.476+02:00That&#39;s right, I sort of assumed that we are de...That&#39;s right, I sort of assumed that we are dealing with hot stream of unpredictable data coming from outside. If it&#39;s not the case, your solution is great, thx!Tomasz Nurkiewiczhttps://www.blogger.com/profile/05938011050162061962noreply@blogger.comtag:blogger.com,1999:blog-6753769565491687768.post-26787651408514711852017-09-11T09:52:04.754+02:002017-09-11T09:52:04.754+02:00There is a small caveat here: events.debounce(), t...There is a small caveat here: events.debounce(), takeUntil(events), merge(events, ...). If events is cold, this will yield multiple subscriptions and multiple runs of the flow, which will likely be unrelated to each other. You sidestepped this issue in the test by using PublishProcessor, perhaps unknowingly, thus the effect doesn&#39;t manifest with hot sources. The fix is trivial: publish(Function)!<br /><br />return events.publish(evt -&gt;<br /> evt.mergeWith(<br /> evt<br /> .debounce(1, SECONDS, clock)<br /> .flatMap(x1 -&gt; Flowable<br /> .interval(0, 1, SECONDS, clock)<br /> .map(e -&gt; ping)<br /> .takeUntil(evt)<br /> )<br />);Dávid Karnokhttps://www.blogger.com/profile/07920580392321059533noreply@blogger.com