New operators: `concatEmptyWith` and `mergeEmptyWith`. #3060

NiteshKant · 2015-07-02T01:28:32Z

As discussed in issue #3037, the primary use of these operators is to be applied to Observable<Void> so that they can be merged and concatenated with an Observable of a different type.

Both these operators raise an error if the source Observable emits any item.

NiteshKant · 2015-07-02T01:35:20Z

Marked the new operators as @Experimental, once the review is complete, I will squash the commits into one.

NiteshKant · 2015-07-14T08:08:23Z

Renamed operators to mergeEmptyWith and concatEmptyWith. Based on offline discussions with @abersnaze and @benjchristensen these names better suit the intent of these operators.

Also, removed the swithMap() + mergeWith/concatWith implementations with custom operators that are much lighter weight for this case (source never emits an item) vis-a-vis concat and merge.

benjchristensen · 2015-07-14T16:23:09Z

Waiting on confirmation of names: #3037 (comment)

roman-mazur · 2015-07-14T16:58:59Z

src/main/java/rx/Observable.java

+     * an item. So, this usually is useful for {@code Observable<Void>} and results in cleaner code as opposed to using
+     * a {@link #cast(Class)}, something like:
+     *
+     {@code Observable.<Void>empty().cast(String.class).concatWith(Observable.just("Hello"))}


nit: no * before the line

abersnaze · 2015-07-14T17:33:55Z

src/main/java/rx/internal/operators/OperatorConcatEmptyWith.java

+            /*Nothing from this producer will ever be requested as we never expect any items to be emitted from the
+            parent. The only thing w.r.t backpressure that is required is to store the requested count from the
+            child till the time the alternate observable is subscribed. So, this producer is just ignored and the
+            configured EmptyProducer on the child stores any buffered requested items till subscription to alternate.*/


How does the subscriber know the producer doesn't emit if we never request it too?

I think if the this should call producer.request(Long.MAX_VALUE);.

If the operator uses ProducerArbiter, the child request will reach the main source and there is no need for extra requests.

I have added producer.request(Long.MAX_VALUE) to the source Observable for both the operators.
There was a thought to send the request to the source only after the subscriber requested anything. Although, this made sense but in this context, it was not making much of a difference but adding complexity of synchronization between request(n) and setProducer(). The argument against being that for a source that never emits an item, should the terminal events be delayed till request(n) or a subscription indicates start of processing.

benjchristensen · 2015-07-14T18:51:54Z

Thanks @akarnokd for the review.

As discussed in issue ReactiveX#3037, the primary use of these operators is to be applied to `Observable<Void>` so that they can be merged and concatenated with an Observable of a different type. Both these operators raise an error if the source Observable emits any item. Review comments

NiteshKant · 2015-07-15T06:36:33Z

Apart from the open discussions, I think I am ready for another round of review.

The travis build failed because of the test rx.BackpressureTests > testMergeAsyncThenObserveOn failing. I don't think my changes have anything to do with that, is that a flaky test?

akarnokd · 2015-08-12T21:50:40Z

src/main/java/rx/internal/operators/OperatorConcatEmptyWith.java

+            }
+
+            if (requestToAlternate) {
+                this.altProducer.request(missedRequested);


You access missedRequested and altProducer outside of their guard.

benjchristensen · 2015-08-12T22:08:13Z

@akarnokd It seems you have open questions/issues on this PR still, is that correct?

akarnokd · 2015-08-12T22:08:33Z

src/main/java/rx/internal/operators/OperatorMergeEmptyWith.java

+        public void onCompleted() {
+            synchronized (this) {
+                if (parentCompleted) {
+                    delegate.onCompleted();


You shouldn't call onXXX methods while holding a lock.

duhh .. that was a bad oversight at my end. Thanks for catching it!

akarnokd · 2015-08-12T22:18:36Z

@benjchristensen Found a few problems with this PR in the meantime.

NiteshKant · 2015-08-13T17:20:36Z

@akarnokd I would wait for your further comments to make all changes together. So, get the review comments coming, they are good :)

abersnaze · 2015-09-30T18:29:48Z

@NiteshKant is this ready to be merged?

sigificent -> significant alreay -> already

…etryWithPredicate

…ed commit) Squashed commits: [c6e43fc] 1.0.15. Beta/Deprecation of Subject state peeking methods. This should give users one release to prepare for the class structure changes.

This came up in a [Stackoverflow](http://stackoverflow.com/questions/32889008/do-operators-instead-of-a-whole-subscriber) answer. If the doOnError's callback or the doOnEach's onError method throws, any non-fatal exception replaced the original error which got lost. This PR will wrap them both into a CompositeException. 2.x note: since Java 8 supports `addSuppressed` all callbacks in this situation either attach to the original exception or the original exception is attached to the callback's exception.

…inal

Slight change to make the distinction between `@Beta` and `@Experimental` explicit and meaningful.

As discussed in issue ReactiveX#3037, the primary use of these operators is to be applied to `Observable<Void>` so that they can be merged and concatenated with an Observable of a different type. Both these operators raise an error if the source Observable emits any item.

Conflicts: src/main/java/rx/internal/operators/OperatorConcatEmptyWith.java src/main/java/rx/internal/operators/OperatorMergeEmptyWith.java

NiteshKant · 2015-10-09T20:15:53Z

I addressed the open comments apart from the suggestion of calling unsubscribe() post onError(), which I do not understand/agree with.

I see that my rebase caused a bazillion other commits to show here, may be I should close this and issue a new PR?

akarnokd · 2015-10-09T20:17:06Z

Please do create a new and clean PR.

NiteshKant · 2015-10-09T21:19:22Z

@akarnokd created #3430 as a replacement for this one.

NiteshKant mentioned this pull request Jul 2, 2015

New experimental operator to switch the types on the completion of empty Observables. #3056

Closed

roman-mazur reviewed Jul 14, 2015
View reviewed changes

NiteshKant changed the title ~~New operators: continueWith and mergeError.~~ New operators: concatEmptyWith and mergeEmptyWith. Jul 14, 2015

abersnaze reviewed Jul 14, 2015
View reviewed changes

NiteshKant force-pushed the 1.x branch from 0a53a04 to 009b3e2 Compare July 15, 2015 06:14

stealthcode mentioned this pull request Jul 20, 2015

Observable.x(ConversionFunc) to allow extensions to Observables #3082

Merged

akarnokd added the Enhancement label Jul 23, 2015

abersnaze mentioned this pull request Jul 25, 2015

new method for ignoreElements then other observable? #3113

Closed

akarnokd reviewed Aug 12, 2015
View reviewed changes

benjchristensen modified the milestone: 1.0.x Aug 28, 2015

stevegury and others added 25 commits October 9, 2015 13:00

Fix typo in a comment inside Observable.subscribe

f9ab970

sigificent -> significant alreay -> already

Make field final and remove unnecessary unboxing in OnSubscribeRedo.R…

d6aabba

…etryWithPredicate

Fix synchronization on non-final field in BufferUntilSubscriber

8859bcf

Fix to a bunch of bugs and issues with AsyncOnSubscribe

3dd46e6

Fix for take() reentrancy bug.

6f73a02

Schedulers shutdown capability.

bf0f3e9

Hiding start(), moved test to compensate.

bf1359f

pull back the Experimental/Beta of the changes until 1.1.x (+1 squash…

730be58

…ed commit) Squashed commits: [c6e43fc] 1.0.15. Beta/Deprecation of Subject state peeking methods. This should give users one release to prepare for the class structure changes.

Added warning to Observable.doOnRequest javadoc.

29775c9

Remove unused private method from CachedObservable and make "state" f…

53c523c

…inal

Safer error handling in BlockingOperatorToFuture

79f2ae7

Remove unnecessary onStart in OperatorGroupBy

c929e0d

Eager concatMap

4cac9f5

Make BlockingOperatorToIterator exert backpressure.

3bafb53

Request data in batches.

782b0e7

BlockingObservable + subscribe methods.

2d0ff46

Update README.md

7aef4f8

Slight change to make the distinction between `@Beta` and `@Experimental` explicit and meaningful.

Renaming Observable#x to Observable#extend

324514a

Add Single.doOnError()

50c2234

Add Single.fromCallable()

6634212

Add Single.doOnSuccess()

22f6ba7

Merge branch '1.x' of https://github.com/Netflix/RxJava into 1.x

8569ff2

Merge branch '1.x' of https://github.com/NiteshKant/RxJava into 1.x

3f6ed32

Conflicts: src/main/java/rx/internal/operators/OperatorConcatEmptyWith.java src/main/java/rx/internal/operators/OperatorMergeEmptyWith.java

NiteshKant closed this Oct 9, 2015

NiteshKant mentioned this pull request Oct 9, 2015

New operators: concatEmptyWith and mergeEmptyWith #3430

Closed

New operators: concatEmptyWith and mergeEmptyWith. #3060

New operators: concatEmptyWith and mergeEmptyWith. #3060

Uh oh!

Conversation

NiteshKant commented Jul 2, 2015

Uh oh!

NiteshKant commented Jul 2, 2015

Uh oh!

NiteshKant commented Jul 14, 2015

Uh oh!

benjchristensen commented Jul 14, 2015

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

benjchristensen commented Jul 14, 2015

Uh oh!

NiteshKant commented Jul 15, 2015

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

benjchristensen commented Aug 12, 2015

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

akarnokd commented Aug 12, 2015

Uh oh!

NiteshKant commented Aug 13, 2015

Uh oh!

abersnaze commented Sep 30, 2015

Uh oh!

NiteshKant commented Oct 9, 2015

Uh oh!

akarnokd commented Oct 9, 2015

Uh oh!

NiteshKant commented Oct 9, 2015

Uh oh!

Uh oh!

New operators: `concatEmptyWith` and `mergeEmptyWith`. #3060

New operators: `concatEmptyWith` and `mergeEmptyWith`. #3060