Not sure what you mean by `takeXXX` wont complete ...

# rx

[cdurham]

Not sure what you mean by takeXXX wont complete but i'd try takeUntil { }