single: add toSingle method to Observable

[LalitMaganti]

• closes Variants of first, single etc. which return Single #3038
• this method allows an observable which is guaranteed to return exactly one
  item to be converted to a Single
• NOTE: the semantics of this function are very similar to that of single
• i.e. errors are passed through, more than one item results in an
  IllegalArgumentException, completion without emission results in a
  NoSuchElementException and exactly one item is passed through the onSuccess
  method of SingleSubscriber

[LalitMaganti]

Would appreciate feedback since this is my first contribution to RxJava! :)

[akarnokd]

Unsubscription chain is not maintained here.

[LalitMaganti]

Just to double check: does this mean I need to add a few checks like
if (singleSubscriber.isUnsubscribed()) {
unsubscribe();
return;
}
in the 4 subscriber methods?

[akarnokd]

See http://akarnokd.blogspot.hu/2015/05/pitfalls-of-operator-implementations.html for a detailed explanation.

[LalitMaganti]

Thanks that helps quite a bit in understanding. However, unlike in your post, I cannot pass the child subscriber to the parent since the child has the SingleSubscriber type which isn't accepted by the Subscriber constructor.

[akarnokd]

But you can add resources to singleSubscriber.

[abersnaze]

There is also a constructor overload that takes the child Subscriber to automatically chain together unsubscriptions.

Sent from my iPhone

On Jun 24, 2015, at 3:54 AM, Lalit Maganti [email protected] wrote:

In src/main/java/rx/internal/operators/OnSubscribeSingle.java:

+/**

• * Allows conversion of an Observable to a Single ensuring that exactly one item is emitted - no more and no less.
• * Also forwards errors as appropriate.
• */
  +public class OnSubscribeSingle implements Single.OnSubscribe {
  +
• private final Observable observable;

• public OnSubscribeSingle(Observable observable) {

•    this.observable = observable;
  

• }

• @OverRide
• public void call(final SingleSubscriber<? super T> singleSubscriber) {

•    observable.subscribe(new Subscriber<T>() {
  

  Just to double check: does this mean I need to add a few checks like
  if (singleSubscriber.isUnsubscribed()) {
  unsubscribe();
  return;
  }
  in the 4 subscriber methods?

—
Reply to this email directly or view it on GitHub.

[LalitMaganti]

As I say though I can't use that overload since the type of the child is SingleSubscriber not Subscriber.

[akarnokd]

Yes.

[benjchristensen]

Looks good. Can you add the license headers to the files please? The block of code to add is shown here: https://github.com/ReactiveX/RxJava/blob/1.x/CONTRIBUTING.md

[LalitMaganti]

Have done so.

[benjchristensen]

Would you mind rebasing to a single commit?

[LalitMaganti]

Also done :)

[benjchristensen]

Thank you @tilal6991 for the contribution and responsiveness on changes.

[LalitMaganti]

Glad to contribute.

I'm planning to look into more tricky issues soon so hopefully this is not the last you'll see of me :)

[davidmoten]

@benjchristensen Shouldn't this use unsafeSubscribe?