Some performance improvements

Author: akarnokd

rxjava-core/src/main/java/rx/subjects/BehaviorSubject.java

@@ -98,6 +98,7 @@ private static <T> BehaviorSubject<T> create(T defaultValue, boolean hasDefault)
     static final class State<T> {
         final AtomicReference<Object> latest = new AtomicReference<Object>();
         final AtomicReference<BehaviorState> observers = new AtomicReference<BehaviorState>(BehaviorState.EMPTY);
+        boolean active = true;
         void set(Object value) {
             this.latest.set(value);
         }
@@ -139,6 +140,7 @@ BehaviorObserver<T>[] next(Object n) {
         }
         BehaviorObserver<T>[] terminate(Object n) {
             set(n);
+            active = false;
             do {
                 BehaviorState oldState = observers.get();
                 if (oldState.terminated) {
@@ -149,9 +151,6 @@ BehaviorObserver<T>[] terminate(Object n) {
                 }
             } while (true);
         }
-        boolean isActive() {
-            return !observers.get().terminated;
-        }
     }
     static final class BehaviorState {
         final boolean terminated;
@@ -241,7 +240,7 @@ protected BehaviorSubject(OnSubscribe<T> onSubscribe, State<T> state) {
     @Override
     public void onCompleted() {
         Object last = state.get();
-        if (last == null || state.isActive()) {
+        if (last == null || state.active) {
             Object n = nl.completed();
             for (BehaviorObserver<T> bo : state.terminate(n)) {
                 bo.emitNext(n);
@@ -252,7 +251,7 @@ public void onCompleted() {
     @Override
     public void onError(Throwable e) {
         Object last = state.get();
-        if (last == null || state.isActive()) {
+        if (last == null || state.active) {
             Object n = nl.error(e);
             for (BehaviorObserver<T> bo : state.terminate(n)) {
                 bo.emitNext(n);
@@ -263,7 +262,7 @@ public void onError(Throwable e) {
     @Override
     public void onNext(T v) {
         Object last = state.get();
-        if (last == null || state.isActive()) {
+        if (last == null || state.active) {
             Object n = nl.next(v);
             for (BehaviorObserver<T> bo : state.next(n)) {
                 bo.emitNext(n);
@@ -284,8 +283,6 @@ private static final class BehaviorObserver<T> {
         boolean emitting;
         /** Guarded by this. */
         List<Object> queue;
-        /** Accessed only from serialized state. */
-        boolean done;
         /* volatile */boolean fastPath;
         public BehaviorObserver(Observer<? super T> actual) {
             this.actual = actual;
@@ -349,10 +346,7 @@ void emitLoop(List<Object> localQueue, Object current) {
             }
         }
         void accept(Object n) {
-            if (n != null && !done) {
-                if (nl.isCompleted(n) || nl.isError(n)) {
-                    done = true;
-                }
+            if (n != null) {
                 nl.accept(actual, n);
             }
         }

rxjava-core/src/test/java/rx/subjects/BehaviorSubjectTest.java

@@ -304,11 +304,12 @@ public void testStartEmpty() {
 
         source.onNext(2);
 
+        verify(o, never()).onError(any(Throwable.class));
+
         inOrder.verify(o).onNext(1);
         inOrder.verify(o).onCompleted();
         inOrder.verifyNoMoreInteractions();
 
-        verify(o, never()).onError(any(Throwable.class));
 
     }
     @Test