2.x: Maybe for lazy Optional

src/main/java/io/reactivex/Flowable.java

@@ -27,6 +27,7 @@
 import io.reactivex.internal.fuseable.*;
 import io.reactivex.internal.operators.completable.CompletableFromPublisher;
 import io.reactivex.internal.operators.flowable.*;
+import io.reactivex.internal.operators.maybe.MaybeFromPublisher;
 import io.reactivex.internal.operators.observable.ObservableFromPublisher;
 import io.reactivex.internal.operators.single.SingleFromPublisher;
 import io.reactivex.internal.schedulers.ImmediateThinScheduler;
@@ -13598,6 +13599,12 @@ public final Single<T> toSingle() {
         return RxJavaPlugins.onAssembly(new SingleFromPublisher<T>(this));
     }
 
+    @BackpressureSupport(BackpressureKind.UNBOUNDED_IN)
+    @SchedulerSupport(SchedulerSupport.NONE)
+    public final Maybe<T> toMaybe() {
+        return new MaybeFromPublisher<T>(this);
+    }
+
     /**
      * Returns a Flowable that emits a list that contains the items emitted by the source Publisher, in a
      * sorted order. Each item emitted by the Publisher must implement {@link Comparable} with respect to all

src/main/java/io/reactivex/MaybeObserver.java

@@ -0,0 +1,26 @@
+/**
+ * Copyright 2016 Netflix, Inc.
+ * 
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in
+ * compliance with the License. You may obtain a copy of the License at
+ * 
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is
+ * distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See
+ * the License for the specific language governing permissions and limitations under the License.
+ */
+
+package io.reactivex;
+
+import io.reactivex.disposables.Disposable;
+
+public interface MaybeObserver<T> {
+    void onSubscribe(Disposable d);
+
+    void onSuccess(T value);
+
+    void onComplete();
+
+    void onError(Throwable e);
+}

src/main/java/io/reactivex/MaybeOperator.java

@@ -0,0 +1,20 @@
+/**
+ * Copyright 2016 Netflix, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in
+ * compliance with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is
+ * distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See
+ * the License for the specific language governing permissions and limitations under the License.
+ */
+
+package io.reactivex;
+
+import io.reactivex.functions.Function;
+
+public interface MaybeOperator<Downstream, Upstream> extends Function<MaybeObserver<? super Downstream>, MaybeObserver<? super Upstream>> {
+
+}

src/main/java/io/reactivex/MaybeSource.java

@@ -0,0 +1,28 @@
+/**
+ * Copyright 2016 Netflix, Inc.
+ * 
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in
+ * compliance with the License. You may obtain a copy of the License at
+ * 
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is
+ * distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See
+ * the License for the specific language governing permissions and limitations under the License.
+ */
+package io.reactivex;
+
+/**
+ * Represents a basic {@link Single} source base interface,
+ * consumable via an {@link SingleObserver}.
+ * <p>
+ * This class also serves the base type for custom operators wrapped into
+ * Single via {@link Single#create(MaybeSource)}.
+ * 
+ * @param <T> the element type
+ * @since 2.0
+ */
+public interface MaybeSource<T> {
+
+    void subscribe(MaybeObserver<? super T> s);
+}

src/main/java/io/reactivex/MaybeTransformer.java

@@ -0,0 +1,20 @@
+/**
+ * Copyright 2016 Netflix, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in
+ * compliance with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is
+ * distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See
+ * the License for the specific language governing permissions and limitations under the License.
+ */
+
+package io.reactivex;
+
+import io.reactivex.functions.Function;
+
+public interface MaybeTransformer<Upstream, Downstream> extends Function<Maybe<Upstream>, MaybeSource<Downstream>> {
+
+}

src/main/java/io/reactivex/functions/Supplier.java

@@ -0,0 +1,26 @@
+/**
+ * Copyright 2016 Netflix, Inc.
+ * 
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in
+ * compliance with the License. You may obtain a copy of the License at
+ * 
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is
+ * distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See
+ * the License for the specific language governing permissions and limitations under the License.
+ */
+
+package io.reactivex.functions;
+
+/**
+ * A functional interface (callback) that returns an Object value.
+ */
+public interface Supplier<R> {
+    /**
+     * Returns an Object value.
+     * @return an Object value
+     * @throws Exception on error
+     */
+    R get() throws Exception; // NOPMD
+}

src/main/java/io/reactivex/internal/operators/maybe/MaybeAmbArray.java

@@ -0,0 +1,87 @@
+/**
+ * Copyright 2016 Netflix, Inc.
+ * 
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in
+ * compliance with the License. You may obtain a copy of the License at
+ * 
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is
+ * distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See
+ * the License for the specific language governing permissions and limitations under the License.
+ */
+
+package io.reactivex.internal.operators.maybe;
+
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import io.reactivex.*;
+import io.reactivex.disposables.*;
+import io.reactivex.plugins.RxJavaPlugins;
+
+public final class MaybeAmbArray<T> extends Maybe<T> {
+
+    final MaybeSource<? extends T>[] sources;
+
+    public MaybeAmbArray(MaybeSource<? extends T>[] sources) {
+        this.sources = sources;
+    }
+
+    @Override
+    protected void subscribeActual(final MaybeObserver<? super T> s) {
+
+        final AtomicBoolean once = new AtomicBoolean();
+        final CompositeDisposable set = new CompositeDisposable();
+        s.onSubscribe(set);
+
+        for (MaybeSource<? extends T> s1 : sources) {
+            if (once.get()) {
+                return;
+            }
+
+            if (s1 == null) {
+                set.dispose();
+                Throwable e = new NullPointerException("One of the sources is null");
+                if (once.compareAndSet(false, true)) {
+                    s.onError(e);
+                } else {
+                    RxJavaPlugins.onError(e);
+                }
+                return;
+            }
+
+            s1.subscribe(new MaybeObserver<T>() {
+
+                @Override
+                public void onSubscribe(Disposable d) {
+                    set.add(d);
+                }
+
+                @Override
+                public void onSuccess(T value) {
+                    if (once.compareAndSet(false, true)) {
+                        s.onSuccess(value);
+                    }
+                }
+
+                @Override
+                public void onComplete() {
+                    if (once.compareAndSet(false, true)) {
+                        s.onComplete();
+                    }
+                }
+
+                @Override
+                public void onError(Throwable e) {
+                    if (once.compareAndSet(false, true)) {
+                        s.onError(e);
+                    } else {
+                        RxJavaPlugins.onError(e);
+                    }
+                }
+
+            });
+        }
+    }
+
+}

src/main/java/io/reactivex/internal/operators/maybe/MaybeAmbIterable.java

@@ -0,0 +1,134 @@
+/**
+ * Copyright 2016 Netflix, Inc.
+ * 
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in
+ * compliance with the License. You may obtain a copy of the License at
+ * 
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is
+ * distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See
+ * the License for the specific language governing permissions and limitations under the License.
+ */
+
+package io.reactivex.internal.operators.maybe;
+
+import java.util.*;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import io.reactivex.*;
+import io.reactivex.disposables.*;
+import io.reactivex.plugins.RxJavaPlugins;
+
+public final class MaybeAmbIterable<T> extends Maybe<T> {
+
+    final Iterable<? extends MaybeSource<? extends T>> sources;
+
+    public MaybeAmbIterable(Iterable<? extends MaybeSource<? extends T>> sources) {
+        this.sources = sources;
+    }
+
+    @Override
+    protected void subscribeActual(final MaybeObserver<? super T> s) {
+        final CompositeDisposable set = new CompositeDisposable();
+        s.onSubscribe(set);
+
+        Iterator<? extends MaybeSource<? extends T>> iterator;
+
+        try {
+            iterator = sources.iterator();
+        } catch (Throwable e) {
+            s.onError(e);
+            return;
+        }
+
+        if (iterator == null) {
+            s.onError(new NullPointerException("The iterator returned is null"));
+            return;
+        }
+
+        final AtomicBoolean once = new AtomicBoolean();
+        int c = 0;
+
+        for (;;) {
+            if (once.get()) {
+                return;
+            }
+
+            boolean b;
+
+            try {
+                b = iterator.hasNext();
+            } catch (Throwable e) {
+                s.onError(e);
+                return;
+            }
+
+            if (once.get()) {
+                return;
+            }
+
+            if (!b) {
+                break;
+            }
+
+            if (once.get()) {
+                return;
+            }
+
+            MaybeSource<? extends T> s1;
+
+            try {
+                s1 = iterator.next();
+            } catch (Throwable e) {
+                set.dispose();
+                s.onError(e);
+                return;
+            }
+
+            if (s1 == null) {
+                set.dispose();
+                s.onError(new NullPointerException("The single source returned by the iterator is null"));
+                return;
+            }
+
+            s1.subscribe(new MaybeObserver<T>() {
+
+                @Override
+                public void onSubscribe(Disposable d) {
+                    set.add(d);
+                }
+
+                @Override
+                public void onSuccess(T value) {
+                    if (once.compareAndSet(false, true)) {
+                        s.onSuccess(value);
+                    }
+                }
+
+                @Override
+                public void onComplete() {
+                    if (once.compareAndSet(false, true)) {
+                        s.onComplete();
+                    }
+                }
+
+                @Override
+                public void onError(Throwable e) {
+                    if (once.compareAndSet(false, true)) {
+                        s.onError(e);
+                    } else {
+                        RxJavaPlugins.onError(e);
+                    }
+                }
+
+            });
+            c++;
+        }
+
+        if (c == 0 && !set.isDisposed()) {
+            s.onError(new NoSuchElementException());
+        }
+    }
+
+}