ObservableCollapser implementation