Migrated from streams to Rx.js