UnsafeSubscribe

Migrate from reflection to unsafeSubscribe as per discussion at #676

Author: benjchristensen

rxjava-core/src/main/java/rx/Observable.java

@@ -61,7 +61,7 @@ public void addActivePlan(ActivePlan0 activePlan) {
     public void subscribe(Object gate) {
         if (subscribed.compareAndSet(false, true)) {
             this.gate = gate;
-            source.materialize().subscribe(this);
+            source.materialize().unsafeSubscribe(this);
         } else {
             throw new IllegalStateException("Can only be subscribed to once.");
         }

rxjava-core/src/main/java/rx/joins/JoinObserver1.java

@@ -72,7 +72,7 @@ public static <T> BlockingObservable<T> from(final Observable<? extends T> o) {
      * "Guideline 6.4: Protect calls to user code from within an operator"
      */
     private Subscription protectivelyWrapAndSubscribe(Subscriber<? super T> observer) {
-        return o.subscribe(new SafeSubscriber<T>(observer));
+        return o.unsafeSubscribe(new SafeSubscriber<T>(observer));
     }
 
     /**

rxjava-core/src/main/java/rx/observables/BlockingObservable.java

@@ -37,7 +37,7 @@ public static <K, T> GroupedObservable<K, T> from(K key, final Observable<T> o)
 
             @Override
             public void call(Subscriber<? super T> s) {
-                o.subscribe(s);
+                o.unsafeSubscribe(s);
             }
         });
     }

rxjava-core/src/main/java/rx/observables/GroupedObservable.java

@@ -91,7 +91,7 @@ public void onNext(T t) {
      * It will then immediately swap itself out for the actual (after a single notification), but since this is now
      * being done on the same producer thread no further buffering will occur.
      */
-    private static class PassThruObserver<T> implements Observer<T> {
+    private static class PassThruObserver<T> extends Subscriber<T> {
 
         private final Observer<? super T> actual;
         // this assumes single threaded synchronous notifications (the Rx contract for a single Observer)
@@ -133,7 +133,7 @@ private void drainIfNeededAndSwitchToActual() {
 
     }
 
-    private static class BufferedObserver<T> implements Observer<T> {
+    private static class BufferedObserver<T> extends Subscriber<T> {
         private final ConcurrentLinkedQueue<Object> buffer = new ConcurrentLinkedQueue<Object>();
 
         @Override

rxjava-core/src/main/java/rx/operators/BufferUntilSubscriber.java

@@ -28,10 +28,12 @@
 import rx.Observer;
 import rx.Scheduler;
 import rx.Scheduler.Inner;
+import rx.Subscriber;
 import rx.Subscription;
 import rx.functions.Action1;
 import rx.functions.Func0;
 import rx.functions.Func1;
+import rx.subscriptions.CompositeSubscription;
 
 /**
  * The base class for operations that break observables into "chunks". Currently buffers and windows.
@@ -408,7 +410,7 @@ public void pushValue(T value) {
      *            The type of object all internal {@link rx.operators.ChunkedOperation.Chunk} objects record.
      *            <C> The type of object being tracked by the {@link Chunk}
      */
-    protected static class ChunkObserver<T, C> implements Observer<T> {
+    protected static class ChunkObserver<T, C> extends Subscriber<T> {
 
         private final Chunks<T, C> chunks;
         private final Observer<? super C> observer;
@@ -492,12 +494,24 @@ public ObservableBasedSingleChunkCreator(NonOverlappingChunks<T, C> chunks, Func
 
         private void listenForChunkEnd() {
             Observable<? extends TClosing> closingObservable = chunkClosingSelector.call();
-            closingObservable.subscribe(new Action1<TClosing>() {
+            closingObservable.unsafeSubscribe(new Subscriber<TClosing>() {
+
+                @Override
+                public void onCompleted() {
+                    
+                }
+
+                @Override
+                public void onError(Throwable e) {
+                    
+                }
+
                 @Override
-                public void call(TClosing closing) {
+                public void onNext(TClosing t) {
                     chunks.emitAndReplaceChunk();
-                    listenForChunkEnd();
+                    listenForChunkEnd();                    
                 }
+                
             });
         }
 
@@ -524,23 +538,47 @@ public void stop() {
      */
     protected static class ObservableBasedMultiChunkCreator<T, C, TOpening, TClosing> implements ChunkCreator {
 
-        private final SafeObservableSubscription subscription = new SafeObservableSubscription();
+        private final CompositeSubscription subscription = new CompositeSubscription();
 
         public ObservableBasedMultiChunkCreator(final OverlappingChunks<T, C> chunks, Observable<? extends TOpening> openings, final Func1<? super TOpening, ? extends Observable<? extends TClosing>> chunkClosingSelector) {
-            subscription.wrap(openings.subscribe(new Action1<TOpening>() {
+            openings.unsafeSubscribe(new Subscriber<TOpening>(subscription) {
+
+                @Override
+                public void onCompleted() {
+                    
+                }
+
                 @Override
-                public void call(TOpening opening) {
+                public void onError(Throwable e) {
+                    
+                }
+
+                @Override
+                public void onNext(TOpening opening) {
                     final Chunk<T, C> chunk = chunks.createChunk();
                     Observable<? extends TClosing> closingObservable = chunkClosingSelector.call(opening);
 
-                    closingObservable.subscribe(new Action1<TClosing>() {
+                    closingObservable.unsafeSubscribe(new Subscriber<TClosing>() {
+
                         @Override
-                        public void call(TClosing closing) {
-                            chunks.emitChunk(chunk);
+                        public void onCompleted() {
+                            
                         }
-                    });
+
+                        @Override
+                        public void onError(Throwable e) {
+                            
+                        }
+
+                        @Override
+                        public void onNext(TClosing t) {
+                            chunks.emitChunk(chunk);                            
+                        }
+                        
+                    });                    
                 }
-            }));
+                
+            });
         }
 
         @Override

rxjava-core/src/main/java/rx/operators/ChunkedOperation.java

@@ -1,6 +1,5 @@
 package rx.operators;
 
-import java.io.ObjectStreamException;
 import java.io.Serializable;
 
 import rx.Notification;

rxjava-core/src/main/java/rx/operators/NotificationLite.java

@@ -20,6 +20,7 @@
 import rx.Observable;
 import rx.Observable.OnSubscribeFunc;
 import rx.Observer;
+import rx.Subscriber;
 import rx.Subscription;
 import rx.functions.Func1;
 
@@ -39,20 +40,18 @@ private static class AllObservable<T> implements OnSubscribeFunc<Boolean> {
         private final Observable<? extends T> sequence;
         private final Func1<? super T, Boolean> predicate;
 
-        private final SafeObservableSubscription subscription = new SafeObservableSubscription();
-
         private AllObservable(Observable<? extends T> sequence, Func1<? super T, Boolean> predicate) {
             this.sequence = sequence;
             this.predicate = predicate;
         }
 
         @Override
         public Subscription onSubscribe(final Observer<? super Boolean> observer) {
-            return subscription.wrap(sequence.subscribe(new AllObserver(observer)));
+            return sequence.unsafeSubscribe(new AllObserver(observer));
 
         }
 
-        private class AllObserver implements Observer<T> {
+        private class AllObserver extends Subscriber<T> {
             private final Observer<? super Boolean> underlying;
 
             private final AtomicBoolean status = new AtomicBoolean(true);
@@ -82,7 +81,7 @@ public void onNext(T args) {
                 if (changed && !result) {
                     underlying.onNext(false);
                     underlying.onCompleted();
-                    subscription.unsubscribe();
+                    unsubscribe();
                 }
             }
         }

rxjava-core/src/main/java/rx/operators/OperationAll.java

@@ -15,13 +15,14 @@
  */
 package rx.operators;
 
-import static rx.functions.Functions.*;
+import static rx.functions.Functions.alwaysTrue;
 
 import java.util.concurrent.atomic.AtomicBoolean;
 
 import rx.Observable;
 import rx.Observable.OnSubscribeFunc;
 import rx.Observer;
+import rx.Subscriber;
 import rx.Subscription;
 import rx.functions.Func1;
 
@@ -81,7 +82,7 @@ private Any(Observable<? extends T> source, Func1<? super T, Boolean> predicate,
         @Override
         public Subscription onSubscribe(final Observer<? super Boolean> observer) {
             final SafeObservableSubscription subscription = new SafeObservableSubscription();
-            return subscription.wrap(source.subscribe(new Observer<T>() {
+            return subscription.wrap(source.unsafeSubscribe(new Subscriber<T>() {
 
                 private final AtomicBoolean hasEmitted = new AtomicBoolean(false);
 

rxjava-core/src/main/java/rx/operators/OperationAny.java

@@ -19,6 +19,7 @@
 import rx.Observable.OnSubscribeFunc;
 import rx.Observer;
 import rx.Subscription;
+import rx.observers.Subscribers;
 
 /**
  * Hides the identity of another observable.
@@ -34,7 +35,7 @@ public OperationAsObservable(Observable<? extends T> source) {
     }
 
     @Override
-    public Subscription onSubscribe(Observer<? super T> t1) {
-        return source.subscribe(t1);
+    public Subscription onSubscribe(final Observer<? super T> t1) {
+        return source.unsafeSubscribe(Subscribers.from(t1));
     }
 }