Observable interface

language-adaptors/rxjava-clojure/src/main/clojure/rx/lang/clojure/interop.clj

@@ -17,12 +17,12 @@
   (let [f-name (gensym "rc")]
     `(let [~f-name ~f]
        (reify
-         ; If they want Func1, give them onSubscribe as well so Observable/create can be
+         ; If they want Func1, give them subscribe as well so Observable/create can be
          ; used seemlessly with rx/fn.
          ~@(if (and (= prefix "rx.util.functions.Func")
                     (some #{1} arities))
-             `(rx.Observable$OnSubscribeFunc
-                (~'onSubscribe [~'this observer#]
+             `(rx.IObservable
+                (~'subscribe [~'this observer#]
                   (~f-name observer#))))
 
          ~@(mapcat (clojure.core/fn [n]
@@ -39,7 +39,7 @@
 
   If the f has the wrong arity, an ArityException will be thrown at runtime.
 
-  This will also implement rx.Observable$OnSubscribeFunc.onSubscribe for use with
+  This will also implement rx.IObservable.subscribe for use with
   Observable/create. In this case, the function must take an Observable as its single
   argument and return a subscription object.
 

language-adaptors/rxjava-clojure/src/test/clojure/rx/lang/clojure/interop_test.clj

@@ -8,7 +8,7 @@
 (deftest test-fn*
   (testing "implements Func0-9"
     (let [f (rx/fn* vector)]
-      (is (instance? rx.Observable$OnSubscribeFunc f))
+      (is (instance? rx.IObservable f))
       (is (instance? rx.util.functions.Func0 f))
       (is (instance? rx.util.functions.Func1 f))
       (is (instance? rx.util.functions.Func2 f))

language-adaptors/rxjava-groovy/src/main/java/rx/lang/groovy/GroovyOnSubscribeFuncWrapper.java

@@ -16,16 +16,16 @@
 package rx.lang.groovy;
 
 import groovy.lang.Closure;
-import rx.Observable.OnSubscribeFunc;
+import rx.IObservable;
 import rx.Observer;
 import rx.Subscription;
 
 /**
- * Concrete wrapper that accepts a {@link Closure} and produces a {@link OnSubscribeFunc}.
+ * Concrete wrapper that accepts a {@link Closure} and produces a {@link IObservable}.
  * 
  * @param <T>
  */
-public class GroovyOnSubscribeFuncWrapper<T> implements OnSubscribeFunc<T> {
+public class GroovyOnSubscribeFuncWrapper<T> implements IObservable<T> {
 
     private final Closure<Subscription> closure;
 
@@ -34,7 +34,7 @@ public GroovyOnSubscribeFuncWrapper(Closure<Subscription> closure) {
     }
 
     @Override
-    public Subscription onSubscribe(Observer<? super T> observer) {
+    public Subscription subscribe(Observer<? super T> observer) {
         return closure.call(observer);
     }
 

language-adaptors/rxjava-groovy/src/main/java/rx/lang/groovy/RxGroovyExtensionModule.java

@@ -33,7 +33,7 @@
 import org.codehaus.groovy.runtime.metaclass.MetaClassRegistryImpl;
 
 import rx.Observable;
-import rx.Observable.OnSubscribeFunc;
+import rx.IObservable;
 import rx.observables.BlockingObservable;
 import rx.util.functions.Action;
 import rx.util.functions.Function;
@@ -109,7 +109,7 @@ public Object invoke(Object object, Object[] arguments) {
                         if (o instanceof Closure) {
                             if (Action.class.isAssignableFrom(m.getParameterTypes()[i])) {
                                 newArgs[i] = new GroovyActionWrapper((Closure) o);
-                            } else if(OnSubscribeFunc.class.isAssignableFrom(m.getParameterTypes()[i])) {
+                            } else if(IObservable.class.isAssignableFrom(m.getParameterTypes()[i])) {
                                 newArgs[i] = new GroovyOnSubscribeFuncWrapper((Closure) o);
                             } else {
                                 newArgs[i] = new GroovyFunctionWrapper((Closure) o);

language-adaptors/rxjava-groovy/src/test/groovy/rx/lang/groovy/ObservableTests.groovy

@@ -31,7 +31,7 @@ import org.mockito.MockitoAnnotations;
 
 import rx.Notification;
 import rx.Observable;
-import rx.Observable.OnSubscribeFunc;
+import rx.IObservable;
 import rx.Observer;
 import rx.Subscription;
 import rx.observables.GroupedObservable;
@@ -483,9 +483,9 @@ def class ObservableTests {
         assertEquals(expected, actual);
     }
 
-    def class AsyncObservable implements OnSubscribeFunc {
+    def class AsyncObservable implements IObservable {
 
-        public Subscription onSubscribe(final Observer<Integer> observer) {
+        public Subscription subscribe(final Observer<Integer> observer) {
             new Thread(new Runnable() {
                 public void run() {
                     try {
@@ -525,14 +525,14 @@ def class ObservableTests {
         public void received(Object o);
     }
 
-    def class TestOnSubscribe implements OnSubscribeFunc<String> {
+    def class TestOnSubscribe implements IObservable<String> {
         private final int count;
 
         public TestOnSubscribe(int count) {
             this.count = count;
         }
 
-        public Subscription onSubscribe(Observer<String> observer) {
+        public Subscription subscribe(Observer<String> observer) {
 
             observer.onNext("hello_" + count);
             observer.onCompleted();

language-adaptors/rxjava-jruby/src/main/resources/rx/lang/jruby/interop.rb

@@ -18,7 +18,7 @@ module Jruby
       class Interop
         WRAPPERS = {
           Java::RxUtilFunctions::Action         => Java::RxLangJruby::JRubyActionWrapper,
-          Java::Rx::Observable::OnSubscribeFunc => false
+          Java::Rx::IObservable => false
         }
 
         WRAPPERS.default = Java::RxLangJruby::JRubyFunctionWrapper

language-adaptors/rxjava-jruby/src/spec/ruby/rx/lang/jruby/interop_spec.rb

@@ -43,7 +43,7 @@
       observable.subscribe(1)
     end
 
-    it "doesn't wrap OnSubscribeFunc arguments" do
+    it "doesn't wrap IObservable arguments" do
       proc = lambda {|observer| observer.onNext("hi")}
       Java::Rx::Observable.__persistent__ = true
       Java::Rx::Observable.should_not_receive(:create_without_wrapping)

language-adaptors/rxjava-kotlin/src/test/kotlin/rx/lang/kotlin/BasicKotlinTests.kt

@@ -29,7 +29,7 @@ import org.junit.Assert.*
 import rx.Notification
 import rx.Subscription
 import kotlin.concurrent.thread
-import rx.Observable.OnSubscribeFunc
+import rx.IObservable
 import rx.lang.kotlin.BasicKotlinTests.AsyncObservable
 
 /**
@@ -344,8 +344,8 @@ public class BasicKotlinTests {
 
     }
 
-    class AsyncObservable : OnSubscribeFunc<Int>{
-        override fun onSubscribe(t1: Observer<in Int>?): Subscription? {
+    class AsyncObservable : IObservable<Int>{
+        override fun subscribe(t1: Observer<in Int>?): Subscription? {
             thread {
                 Thread.sleep(50)
                 t1!!.onNext(1)
@@ -357,8 +357,8 @@ public class BasicKotlinTests {
         }
     }
 
-    class TestOnSubscribe(val count: Int) : OnSubscribeFunc<String>{
-        override fun onSubscribe(t1: Observer<in String>?): Subscription? {
+    class TestOnSubscribe(val count: Int) : IObservable<String>{
+        override fun subscribe(t1: Observer<in String>?): Subscription? {
             t1!!.onNext("hello_$count")
             t1.onCompleted()
             return Subscription { }

language-adaptors/rxjava-scala/src/main/scala/rx/lang/scala/ImplicitFunctionConversions.scala

@@ -50,8 +50,8 @@ object ImplicitFunctionConversions {
     }
 
   implicit def scalaFunction1ToOnSubscribeFunc[T](f: rx.lang.scala.Observer[T] => Subscription) =
-    new rx.Observable.OnSubscribeFunc[T] {
-      def onSubscribe(obs: rx.Observer[_ >: T]): rx.Subscription = {
+    new rx.IObservable[T] {
+      def subscribe(obs: rx.Observer[_ >: T]): rx.Subscription = {
         f(obs)
       }
     }

language-adaptors/rxjava-scala/src/main/scala/rx/lang/scala/Observable.scala

@@ -17,7 +17,7 @@
 package rx.lang.scala
 
 import rx.util.functions.FuncN
-import rx.Observable.OnSubscribeFunc
+import rx.IObservable
 import rx.lang.scala.observables.ConnectableObservable
 
 
@@ -2081,8 +2081,8 @@ object Observable {
    *         an Observable that, when an [[rx.lang.scala.Observer]] subscribes to it, will execute the given function.
    */
   def create[T](func: Observer[T] => Subscription): Observable[T] = {
-    toScalaObservable[T](rx.Observable.create(new OnSubscribeFunc[T] {
-      def onSubscribe(t1: rx.Observer[_ >: T]): rx.Subscription = {
+    toScalaObservable[T](rx.Observable.create(new IObservable[T] {
+      def subscribe(t1: rx.Observer[_ >: T]): rx.Subscription = {
         func(Observer(t1))
       }
     }))

language-adaptors/rxjava-scala/src/test/scala/rx/lang/scala/CompletenessTest.scala

@@ -111,7 +111,7 @@ class CompletenessTest extends JUnitSuite {
       "averageDoubles(Observable[Double])" -> averageProblem,
       "averageFloats(Observable[Float])" -> averageProblem,
       "averageLongs(Observable[Long])" -> averageProblem,
-      "create(OnSubscribeFunc[T])" -> "apply(Observer[T] => Subscription)",
+      "create(IObservable[T])" -> "apply(Observer[T] => Subscription)",
       "combineLatest(Observable[_ <: T1], Observable[_ <: T2], Func2[_ >: T1, _ >: T2, _ <: R])" -> "combineLatest(Observable[U])",
       "concat(Observable[_ <: Observable[_ <: T]])" -> "concat(<:<[Observable[T], Observable[Observable[U]]])",
       "defer(Func0[_ <: Observable[_ <: T]])" -> "defer(=> Observable[T])",

rxjava-contrib/rxjava-android/src/main/java/rx/operators/OperationObserveFromAndroidComponent.java

@@ -62,7 +62,7 @@ public static <T> Observable<T> observeFromAndroidComponent(Observable<T> source
         return Observable.create(new OnSubscribeBase<T, Activity>(source, activity));
     }
 
-    private static class OnSubscribeBase<T, AndroidComponent> implements Observable.OnSubscribeFunc<T> {
+    private static class OnSubscribeBase<T, AndroidComponent> implements IObservable<T> {
 
         private static final String LOG_TAG = "AndroidObserver";
 
@@ -88,7 +88,7 @@ protected boolean isComponentValid(AndroidComponent component) {
         }
 
         @Override
-        public Subscription onSubscribe(Observer<? super T> observer) {
+        public Subscription subscribe(Observer<? super T> observer) {
             assertUiThread();
             observerRef = observer;
             final Subscription sourceSub = source.observeOn(AndroidSchedulers.mainThread()).subscribe(new Observer<T>() {
@@ -232,7 +232,7 @@ public void itDropsOnNextOnCompletedSequenceIfTargetComponentIsGone() throws Thr
             PublishSubject<Integer> source = PublishSubject.create();
 
             final OnSubscribeFragment<Integer> operator = new OnSubscribeFragment<Integer>(source, mockFragment);
-            operator.onSubscribe(mockObserver);
+            operator.subscribe(mockObserver);
 
             source.onNext(1);
             releaseComponentRef(operator);
@@ -250,7 +250,7 @@ public void itDropsOnErrorIfTargetComponentIsGone() throws Throwable {
             PublishSubject<Integer> source = PublishSubject.create();
 
             final OnSubscribeFragment<Integer> operator = new OnSubscribeFragment<Integer>(source, mockFragment);
-            operator.onSubscribe(mockObserver);
+            operator.subscribe(mockObserver);
 
             source.onNext(1);
             releaseComponentRef(operator);

rxjava-contrib/rxjava-apache-http/src/main/java/rx/apache/http/ObservableHttp.java

@@ -23,7 +23,7 @@
 import org.apache.http.nio.protocol.HttpAsyncRequestProducer;
 
 import rx.Observable;
-import rx.Observable.OnSubscribeFunc;
+import rx.IObservable;
 import rx.Observer;
 import rx.Subscription;
 import rx.apache.http.consumers.ResponseConsumerDelegate;
@@ -78,22 +78,22 @@
  */
 public class ObservableHttp<T> {
 
-    private final OnSubscribeFunc<T> onSubscribe;
+    private final IObservable<T> onSubscribe;
 
-    private ObservableHttp(OnSubscribeFunc<T> onSubscribe) {
+    private ObservableHttp(IObservable<T> onSubscribe) {
         this.onSubscribe = onSubscribe;
     }
 
-    private static <T> ObservableHttp<T> create(OnSubscribeFunc<T> onSubscribe) {
+    private static <T> ObservableHttp<T> create(IObservable<T> onSubscribe) {
         return new ObservableHttp<T>(onSubscribe);
     }
 
     public Observable<T> toObservable() {
-        return Observable.create(new OnSubscribeFunc<T>() {
+        return Observable.create(new IObservable<T>() {
 
             @Override
-            public Subscription onSubscribe(Observer<? super T> observer) {
-                return onSubscribe.onSubscribe(observer);
+            public Subscription subscribe(Observer<? super T> observer) {
+                return onSubscribe.subscribe(observer);
             }
         });
     }
@@ -135,10 +135,10 @@ public static ObservableHttp<ObservableHttpResponse> createGet(String uri, final
      */
     public static ObservableHttp<ObservableHttpResponse> createRequest(final HttpAsyncRequestProducer requestProducer, final HttpAsyncClient client) {
 
-        return ObservableHttp.create(new OnSubscribeFunc<ObservableHttpResponse>() {
+        return ObservableHttp.create(new IObservable<ObservableHttpResponse>() {
 
             @Override
-            public Subscription onSubscribe(final Observer<? super ObservableHttpResponse> observer) {
+            public Subscription subscribe(final Observer<? super ObservableHttpResponse> observer) {
 
                 final CompositeSubscription parentSubscription = new CompositeSubscription();
 

rxjava-contrib/rxjava-apache-http/src/main/java/rx/apache/http/consumers/ResponseConsumerBasic.java

@@ -27,7 +27,7 @@
 import org.apache.http.protocol.HttpContext;
 
 import rx.Observable;
-import rx.Observable.OnSubscribeFunc;
+import rx.IObservable;
 import rx.Observer;
 import rx.Subscription;
 import rx.apache.http.ObservableHttpResponse;
@@ -68,10 +68,10 @@ public void _onEntityEnclosed(HttpEntity entity, ContentType contentType) throws
     public HttpResponse _buildResult(HttpContext context) throws Exception {
         final HttpResponse response = buildResult(context);
 
-        Observable<byte[]> contentObservable = Observable.create(new OnSubscribeFunc<byte[]>() {
+        Observable<byte[]> contentObservable = Observable.create(new IObservable<byte[]>() {
 
             @Override
-            public Subscription onSubscribe(Observer<? super byte[]> o) {
+            public Subscription subscribe(Observer<? super byte[]> o) {
                 long length = response.getEntity().getContentLength();
                 if (length > Integer.MAX_VALUE) {
                     o.onError(new IllegalStateException("Content Length too large for a byte[] => " + length));

rxjava-contrib/rxjava-apache-http/src/main/java/rx/apache/http/consumers/ResponseConsumerEventStream.java

@@ -29,7 +29,7 @@
 import org.apache.http.protocol.HttpContext;
 
 import rx.Observable;
-import rx.Observable.OnSubscribeFunc;
+import rx.IObservable;
 import rx.Observer;
 import rx.Subscription;
 import rx.apache.http.ObservableHttpResponse;
@@ -81,10 +81,10 @@ protected void onByteReceived(ByteBuffer buf, IOControl ioctrl) throws IOExcepti
     protected void onResponseReceived(HttpResponse response) throws HttpException, IOException {
 
         // wrap the contentSubject so we can chain the Subscription between parent and child
-        Observable<byte[]> contentObservable = Observable.create(new OnSubscribeFunc<byte[]>() {
+        Observable<byte[]> contentObservable = Observable.create(new IObservable<byte[]>() {
 
             @Override
-            public Subscription onSubscribe(Observer<? super byte[]> observer) {
+            public Subscription subscribe(Observer<? super byte[]> observer) {
                 parentSubscription.add(contentSubject.subscribe(observer));
                 return parentSubscription;
             }