Created
October 7, 2015 14:11
-
-
Save zenparsing/0d18579ed0c76ab70db7 to your computer and use it in GitHub Desktop.
Switch using cancellation functions
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
// For a nested stream, emits the elements of the inner stream contained within the | |
// most recent outer stream | |
function switch(stream, n) { | |
return new Observable(sink => { | |
let inner = []; | |
let outer = stream.subscribe({ | |
next(value) { | |
while (inner.length >= n) | |
inner.shift()(); | |
let observer = { | |
cancel: null, | |
next: x => sink.next(x), | |
error: x => sink.error(x), | |
complete: _=> { inner.splice(inner.indexOf(this.cancel), 1) }, | |
}; | |
observer.cancel = value.subscribe(observer); | |
}, | |
error: x => sink.error(x), | |
complete: x => sink.complete(x), | |
}); | |
return _=> { | |
while (inner.length > 0) | |
inner.shift()(); | |
outer(); | |
}; | |
}); | |
} |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
It took me a long while to realize that
inner.shift()()
is invoking the unsubscribe function. As a javascript newb,inner.shift().unsubscribe()
would be much clearer for me to infer what that second function call does, but this is only an argument if you're striving to promote clarity for newcomers to Observable that like to dig through source with less 'technical debt', if you will.That said, I think your intent was to call
inner.shift().cancel()
--which achieves the clarity goal above--since you took the time to stash theunsubscribe
callback inobserver.cancel
, but never pushed anything intoinner
.