Add Publisher.zip #2690
Add Publisher.zip #2690
Conversation
Scottmitch
requested review from
daschl,
mgodave,
bryce-anderson,
tkountis and
idelpivnitskiy
Scottmitch
force-pushed
the
publisher_zip
branch
from
ba7bf04
to
e2c6ac2
Compare
Scottmitch
force-pushed
the
publisher_zip
branch
from
e2c6ac2
to
1f898a3
Compare
Motivation: The zip operator helps combine results from multiple Publishers into a single Publisher.
Scottmitch
force-pushed
the
publisher_zip
branch
from
1f898a3
to
7c7deae
Compare
| /** | ||
| * Create a new {@link Publisher} that emits the results of a specified zipper {@link BiFunction} to items emitted | ||
| * by {@code this} and {@code other}. If any of the {@link Publisher}s terminate with an error, the returned | ||
| * {@link Publisher} will wait for termination till all the other {@link Publisher}s have been subscribed and |
There was a problem hiding this comment.
| * {@link Publisher} will wait for termination till all the other {@link Publisher}s have been subscribed and | |
| * {@link Publisher} will wait for termination until all the other {@link Publisher}s have been subscribed and |
same with the others
| for (int i = 63; i >= zipperArity; --i) { | ||
| nonEmptyQueueIndexes |= (1L << i); | ||
| } |
There was a problem hiding this comment.
Alternatively, you could clear the bits from a starting ALL_NON_EMPTY_MASK as you do when polling elements from queues, since the arity will almost always be low single digits.
| if (maxOutstandingDemand <= 0) { | ||
| throw new IllegalArgumentException("maxOutstandingDemand: " + maxOutstandingDemand + " (expected>0)"); | ||
| } |
There was a problem hiding this comment.
In my minds eye we should return an error if the concurrency exceeds the MAX_CONCURRENCY so it doesn't happen lazily.
| public void onComplete() { | ||
| List<Integer> nonEmptyIndexes = new ArrayList<>(); | ||
| for (int i = 0; i < array.length; ++i) { | ||
| if ((nonEmptyQueueIndexes & (1L << i)) != 0) { | ||
| nonEmptyIndexes.add(i); | ||
| } | ||
| } | ||
| if (nonEmptyIndexes.isEmpty()) { | ||
| subscriber.onComplete(); | ||
| } else { | ||
| StringBuilder sb = new StringBuilder(20 + 68 + nonEmptyIndexes.size() * 4); | ||
| sb.append("Publisher indexes: ["); | ||
| Iterator<Integer> itr = nonEmptyIndexes.iterator(); | ||
| sb.append(itr.next()); // safe to call next(), already checked is not empty. | ||
| while (itr.hasNext()) { | ||
| sb.append(", ").append(itr.next()); | ||
| } | ||
| sb.append("] had onNext signals queued when onComplete terminal signal received"); | ||
| subscriber.onError(new IllegalStateException(sb.toString())); | ||
| } | ||
| } |
There was a problem hiding this comment.
I don't think this is the same behavior described in https://reactivex.io/documentation/operators/zip.html. It suggests that its okay to have extra signals emitted and they are simply dropped which is the behavior of most collection zip operators that I've seen. At a minimum, the behavior should be documented so users know they should have publishers of all the same length, which I think could be non-trivial in some cases.
Along these same lines, I don't think this implementation will ever terminate if one of the suppliers is an infinite stream. It seems pretty reasonable to have one source that is an infinite stream an the join it with another finite stream and expect a finite result.
Motivation:
The zip operator helps combine results from multiple Publishers into a single Publisher.