Subject state information methods & bounded ReplaySubject termination

fix

Author: akarnokd

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

@@ -176,6 +176,23 @@ public boolean isError(Object n) {
         return n instanceof OnErrorSentinel;
     }
 
+    /**
+     * Indicates whether or not the lite notification represents a wrapped {@code null} {@code onNext} event.
+     * @param n the lite notification
+     * @return {@code true} if {@code n} represents a wrapped {@code null} {@code onNext} event, {@code false} otherwise
+     */
+    public boolean isNull(Object n) {
+        return n == ON_NEXT_NULL_SENTINEL;
+    }
+
+    /**
+     * Indicates whether or not the lite notification represents an {@code onNext} event.
+     * @param n the lite notification
+     * @return {@code true} if {@code n} represents an {@code onNext} event, {@code false} otherwise
+     */
+    public boolean isNext(Object n) {
+        return n != null && !isError(n) && !isCompleted(n);
+    }
     /**
      * Indicates which variety a particular lite notification is. If you need something more complex than
      * simply calling the right method on an {@link Observer} then you can use this method to get the

src/main/java/rx/subjects/AsyncSubject.java

@@ -140,4 +140,61 @@ public void onNext(T v) {
     public boolean hasObservers() {
         return state.observers().length > 0;
     }
+    /**
+     * Check if the Subject has a value.
+     * <p>Use the {@link #getValue()} method to retrieve such a value.
+     * <p>Note that unless {@link #hasCompleted()} or {@link #hasThrowable()} returns true, the value
+     * retrieved by {@code getValue()} may get outdated.
+     * @return true if and only if the subject has some value but not an error
+     */
+    public boolean hasValue() {
+        Object v = lastValue;
+        Object o = state.get();
+        return !nl.isError(o) && nl.isNext(v);
+    }
+    /**
+     * Check if the Subject has terminated with an exception.
+     * @return true if the subject has received a throwable through {@code onError}.
+     */
+    public boolean hasThrowable() {
+        Object o = state.get();
+        return nl.isError(o);
+    }
+    /**
+     * Check if the Subject has terminated normally.
+     * @return true if the subject completed normally via {@code onCompleted()}
+     */
+    public boolean hasCompleted() {
+        Object o = state.get();
+        return o != null && !nl.isError(o);
+    }
+    /**
+     * Returns the current value of the Subject if there is such a value and
+     * the subject hasn't terminated with an exception.
+     * <p>The can return {@code null} for various reasons. Use {@link #hasValue()}, {@link #hasThrowable()}
+     * and {@link #hasCompleted()} to determine if such {@code null} is a valid value, there was an
+     * exception or the Subject terminated without receiving any value. 
+     * @return the current value or {@code null} if the Subject doesn't have a value,
+     * has terminated with an exception or has an actual {@code null} as a value.
+     */
+    public T getValue() {
+        Object v = lastValue;
+        Object o = state.get();
+        if (!nl.isError(o) && nl.isNext(v)) {
+            return nl.getValue(v);
+        }
+        return null;
+    }
+    /**
+     * Returns the Throwable that terminated the Subject.
+     * @return the Throwable that terminated the Subject or {@code null} if the
+     * subject hasn't terminated yet or it terminated normally.
+     */
+    public Throwable getThrowable() {
+        Object o = state.get();
+        if (nl.isError(o)) {
+            return nl.getError(o);
+        }
+        return null;
+    }
 }

src/main/java/rx/subjects/BehaviorSubject.java

@@ -176,4 +176,59 @@ public void onNext(T v) {
     public boolean hasObservers() {
         return state.observers().length > 0;
     }
+    /**
+     * Check if the Subject has a value.
+     * <p>Use the {@link #getValue()} method to retrieve such a value.
+     * <p>Note that unless {@link #hasCompleted()} or {@link #hasThrowable()} returns true, the value
+     * retrieved by {@code getValue()} may get outdated.
+     * @return true if and only if the subject has some value and hasn't terminated yet.
+     */
+    public boolean hasValue() {
+        Object o = state.get();
+        return nl.isNext(o);
+    }
+    /**
+     * Check if the Subject has terminated with an exception.
+     * @return true if the subject has received a throwable through {@code onError}.
+     */
+    public boolean hasThrowable() {
+        Object o = state.get();
+        return nl.isError(o);
+    }
+    /**
+     * Check if the Subject has terminated normally.
+     * @return true if the subject completed normally via {@code onCompleted()}
+     */
+    public boolean hasCompleted() {
+        Object o = state.get();
+        return nl.isCompleted(o);
+    }
+    /**
+     * Returns the current value of the Subject if there is such a value and
+     * the subject hasn't terminated yet.
+     * <p>The can return {@code null} for various reasons. Use {@link #hasValue()}, {@link #hasThrowable()}
+     * and {@link #hasCompleted()} to determine if such {@code null} is a valid value, there was an
+     * exception or the Subject terminated (with or without receiving any value). 
+     * @return the current value or {@code null} if the Subject doesn't have a value,
+     * has terminated or has an actual {@code null} as a valid value.
+     */
+    public T getValue() {
+        Object o = state.get();
+        if (nl.isNext(o)) {
+            return nl.getValue(o);
+        }
+        return null;
+    }
+    /**
+     * Returns the Throwable that terminated the Subject.
+     * @return the Throwable that terminated the Subject or {@code null} if the
+     * subject hasn't terminated yet or it terminated normally.
+     */
+    public Throwable getThrowable() {
+        Object o = state.get();
+        if (nl.isError(o)) {
+            return nl.getError(o);
+        }
+        return null;
+    }
 }

src/main/java/rx/subjects/PublishSubject.java

@@ -126,4 +126,33 @@ public void onNext(T v) {
     public boolean hasObservers() {
         return state.observers().length > 0;
     }
+    
+    /**
+     * Check if the Subject has terminated with an exception.
+     * @return true if the subject has received a throwable through {@code onError}.
+     */
+    public boolean hasThrowable() {
+        Object o = state.get();
+        return nl.isError(o);
+    }
+    /**
+     * Check if the Subject has terminated normally.
+     * @return true if the subject completed normally via {@code onCompleted}
+     */
+    public boolean hasCompleted() {
+        Object o = state.get();
+        return o != null && !nl.isError(o);
+    }
+    /**
+     * Returns the Throwable that terminated the Subject.
+     * @return the Throwable that terminated the Subject or {@code null} if the
+     * subject hasn't terminated yet or it terminated normally.
+     */
+    public Throwable getThrowable() {
+        Object o = state.get();
+        if (nl.isError(o)) {
+            return nl.getError(o);
+        }
+        return null;
+    }
 }