2.x: remove Try+Optional, introduce Notification (#4370)

Author: akarnokd

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

@@ -6816,7 +6816,7 @@ public final Flowable<T> delaySubscription(long delay, TimeUnit unit, Scheduler
     @SchedulerSupport(SchedulerSupport.NONE)
     public final <T2> Flowable<T2> dematerialize() {
         @SuppressWarnings("unchecked")
-        Flowable<Try<Optional<T2>>> m = (Flowable<Try<Optional<T2>>>)this;
+        Flowable<Notification<T2>> m = (Flowable<Notification<T2>>)this;
         return new FlowableDematerialize<T2>(m);
     }
 
@@ -7108,7 +7108,7 @@ private Flowable<T> doOnEach(Consumer<? super T> onNext, Consumer<? super Throwa
      */
     @BackpressureSupport(BackpressureKind.PASS_THROUGH)
     @SchedulerSupport(SchedulerSupport.NONE)
-    public final Flowable<T> doOnEach(final Consumer<? super Try<Optional<T>>> onNotification) {
+    public final Flowable<T> doOnEach(final Consumer<? super Notification<T>> onNotification) {
         Objects.requireNonNull(onNotification, "consumer is null");
         return doOnEach(
                 Functions.notificationOnNext(onNotification),
@@ -8670,7 +8670,7 @@ public final <R> Flowable<R> map(Function<? super T, ? extends R> mapper) {
      */
     @BackpressureSupport(BackpressureKind.FULL)
     @SchedulerSupport(SchedulerSupport.NONE)
-    public final Flowable<Try<Optional<T>>> materialize() {
+    public final Flowable<Notification<T>> materialize() {
         return new FlowableMaterialize<T>(this);
     }
 

src/main/java/io/reactivex/Notification.java

@@ -14,54 +14,142 @@
 package io.reactivex;
 
 import io.reactivex.internal.functions.Objects;
+import io.reactivex.internal.util.NotificationLite;
 
 /**
- * Utility class to help construct notification objects.
+ * Represents the reactive signal types: onNext, onError and onComplete and
+ * holds their parameter values (a value, a Throwable, nothing).
+ * @param <T> the value type
  */
-public final class Notification {
-    static final Try<Optional<Object>> COMPLETE = Try.ofValue(Optional.<Object>empty());
+public final class Notification<T> {
 
-    private Notification() {
-        throw new IllegalStateException();
+    final Object value;
+    
+    /** Not meant to be implemented externally. */
+    private Notification(Object value) {
+        this.value = value;
     }
 
-    @SuppressWarnings({ "rawtypes", "unchecked" })
-    public static <T> Try<Optional<T>> complete() {
-        return (Try)COMPLETE;
+    /**
+     * Returns true if this notification is an onComplete signal.
+     * @return true if this notification is an onComplete signal
+     */
+    public boolean isOnComplete() {
+        return value == null;
     }
 
-    @SuppressWarnings({ "rawtypes", "unchecked" })
-    public static <T> Try<Optional<T>> error(Throwable e) {
-        return (Try)Try.ofError(e);
+    /**
+     * Returns true if this notification is an onError signal and
+     * {@link #getError()} returns the contained Throwable.
+     * @return true if this notification is an onError signal
+     * @see #getError()
+     */
+    public boolean isOnError() {
+        return NotificationLite.isError(value);
     }
 
-    public static <T> Try<Optional<T>> next(T value) {
-        Objects.requireNonNull(value, "value is null"); // TODO this coud instead return an error of NPE
-        return Try.ofValue(Optional.of(value));
+    /**
+     * Returns true if this notification is an onNext signal and
+     * {@link #getValue()} returns the contained value.
+     * @return true if this notification is an onNext signal
+     * @see #getValue()
+     */
+    public boolean isOnNext() {
+        Object o = value;
+        return o != null && !NotificationLite.isError(o);
     }
 
-    public static <T> boolean isNext(Try<Optional<T>> notification) {
-        if (notification.hasValue()) {
-            return notification.value().isPresent();
+    /**
+     * Returns the contained value if this notification is an onNext
+     * signal, null otherwise.
+     * @return the value contained or null
+     * @see #isOnNext()
+     */
+    @SuppressWarnings("unchecked")
+    public T getValue() {
+        Object o = value;
+        if (o != null && !NotificationLite.isError(o)) {
+            return (T)value;
         }
-        return false;
+        return null;
+    }
+    
+    /**
+     * Returns the container Throwable error if this notification is an onError
+     * signal, null otherwise.
+     * @return the Throwable error contained or null
+     * @see #isOnError()
+     */
+    public Throwable getError() {
+        Object o = value;
+        if (NotificationLite.isError(o)) {
+            return NotificationLite.getError(o);
+        }
+        return null;
     }
 
-    public static <T> boolean isComplete(Try<Optional<T>> notification) {
-        if (notification.hasValue()) {
-            return !notification.value().isPresent();
+    @Override
+    public boolean equals(Object obj) {
+        if (obj instanceof Notification) {
+            Notification<?> n = (Notification<?>) obj;
+            return Objects.equals(value, n.value);
         }
         return false;
     }
 
-    public static <T> boolean isError(Try<Optional<T>> notification) {
-        return notification.hasError();
+    @Override
+    public int hashCode() {
+        Object o = value;
+        return o != null ? o.hashCode() : 0;
     }
 
-    public static <T> T getValue(Try<Optional<T>> notification) {
-        if (notification.hasValue()) {
-            return notification.value.get();
+    @Override
+    public String toString() {
+        Object o = value;
+        if (o == null) {
+            return "OnCompleteNotification";
         }
-        return null;
+        if (NotificationLite.isError(o)) {
+            return "OnErrorNotification[" + NotificationLite.getError(o) + "]";
+        }
+        return "OnNextNotification[" + value + "]";
+    }
+    
+    /**
+     * Constructs an onNext notification containing the given value.
+     * @param <T> the value type
+     * @param value the value to carry around in the notification, not null
+     * @return the new Notification instance
+     * @throws NullPointerException if value is null
+     */
+    public static <T> Notification<T> createOnNext(T value) {
+        Objects.requireNonNull(value, "value is null");
+        return new Notification<T>(value);
+    }
+    
+    /**
+     * Constructs an onError notification containing the error.
+     * @param <T> the value type
+     * @param error the error Throwable to carry around in the notification, not null
+     * @return the new Notification instance
+     * @throws NullPointerException if error is null
+     */
+    public static <T> Notification<T> createOnError(Throwable error) {
+        Objects.requireNonNull(error, "error is null");
+        return new Notification<T>(NotificationLite.error(error));
+    }
+    
+    /**
+     * Returns the empty and stateless shared instance of a notification representing
+     * an onComplete signal.
+     * @param <T> the target value type
+     * @return the shared Notification instance representing an onComplete signal
+     */
+    @SuppressWarnings("unchecked")
+    public static <T> Notification<T> createOnComplete() {
+        return (Notification<T>)COMPLETE;
     }
+    
+    /** The singleton instance for createOnComplete. */
+    static final Notification<Object> COMPLETE = new Notification<Object>(null);
 }

src/main/java/io/reactivex/Observable.java

@@ -6043,7 +6043,7 @@ public final Observable<T> delaySubscription(long delay, TimeUnit unit, Schedule
     @SchedulerSupport(SchedulerSupport.NONE)
     public final <T2> Observable<T2> dematerialize() {
         @SuppressWarnings("unchecked")
-        Observable<Try<Optional<T2>>> m = (Observable<Try<Optional<T2>>>)this;
+        Observable<Notification<T2>> m = (Observable<Notification<T2>>)this;
         return new ObservableDematerialize<T2>(m);
     }
 
@@ -6288,7 +6288,7 @@ private Observable<T> doOnEach(Consumer<? super T> onNext, Consumer<? super Thro
      * @see <a href="http://reactivex.io/documentation/operators/do.html">ReactiveX operators documentation: Do</a>
      */
     @SchedulerSupport(SchedulerSupport.NONE)
-    public final Observable<T> doOnEach(final Consumer<? super Try<Optional<T>>> onNotification) {
+    public final Observable<T> doOnEach(final Consumer<? super Notification<T>> onNotification) {
         Objects.requireNonNull(onNotification, "consumer is null");
         return doOnEach(
                 Functions.notificationOnNext(onNotification),
@@ -7576,7 +7576,7 @@ public final <R> Observable<R> map(Function<? super T, ? extends R> mapper) {
      * @see <a href="http://reactivex.io/documentation/operators/materialize-dematerialize.html">ReactiveX operators documentation: Materialize</a>
      */
     @SchedulerSupport(SchedulerSupport.NONE)
-    public final Observable<Try<Optional<T>>> materialize() {
+    public final Observable<Notification<T>> materialize() {
         return new ObservableMaterialize<T>(this);
     }