Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
Observable zip with Various Return Types
function zip(list) {
return new Observable(sink => {
function trySend() {
// If every buffer has at least one element, then send an array
// containing every first element and remove those elements from the
// buffer
if (streams.every(s => s.buffer.length > 0))
sink.next(streams.map(s => s.buffer.shift()));
tryComplete();
}
function tryComplete() {
// If there is a stream which is completed and whose buffer is empty,
// then send a complete signal to the sink
if (streams.some(s => s.completed && s.buffer.length === 0))
sink.complete();
}
let streams = Array.from(list, observable => {
let stream = { buffer: [], cancel: null, completed: false };
stream.cancel = observable.subscribe({
next(v) {
if (stream.buffer.push(v) === 1)
trySend();
},
error(e) {
return sink.error(e);
},
complete() {
stream.completed = true;
tryComplete();
},
});
return stream;
});
return _=> {
// On cleanup, cancel each subscription
for (let stream of streams)
stream.cancel();
};
});
}
function zip(list) {
return new Observable(sink => {
function trySend() {
// If every buffer has at least one element, then send an array
// containing every first element and remove those elements from the
// buffer
if (streams.every(s => s.buffer.length > 0))
sink.next(streams.map(s => s.buffer.shift()));
tryComplete();
}
function tryComplete() {
// If there is a stream which is completed and whose buffer is empty,
// then send a complete signal to the sink
if (streams.some(s => s.subscription.isUnsubscribed() && s.buffer.length === 0))
sink.complete();
}
let streams = Array.from(list, observable => {
let stream = { buffer: [], subscription: null };
stream.subscription = observable.subscribe({
next(v) {
if (stream.buffer.push(v) === 1)
trySend();
},
error(e) {
return sink.error(e);
},
complete() {
tryComplete();
},
});
return stream;
});
return _=> {
// On cleanup, cancel each subscription
for (let stream of streams)
stream.subscription.unsubscribe();
};
});
}
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.