I should also point out that this essentially constitutes a memory leak unless you know the stream isn't infinite, which is probably why there's no `distinct` method on Observables to begin with.