update rx dependency to 1.1.1

build.gradle

@@ -7,7 +7,7 @@ apply plugin: 'rxjava-project'
 apply plugin: 'java'
 
 dependencies {
-    compile 'io.reactivex:rxjava:1.0.12'
+    compile 'io.reactivex:rxjava:1.1.1'
     testCompile 'junit:junit-dep:4.10'
     testCompile 'org.mockito:mockito-core:1.8.5'
 }

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

@@ -1,13 +1,13 @@
 package rx.internal.operators;
 
+import rx.Observer;
+import rx.observables.SyncOnSubscribe;
+
 import java.io.IOException;
 import java.io.InputStream;
 import java.util.Arrays;
 
-import rx.Subscriber;
-import rx.observables.AbstractOnSubscribe;
-
-public final class OnSubscribeInputStream extends AbstractOnSubscribe<byte[], InputStream> {
+public final class OnSubscribeInputStream extends SyncOnSubscribe<InputStream, byte[]> {
 
     private final InputStream is;
     private final int size;
@@ -18,25 +18,24 @@ public OnSubscribeInputStream(InputStream is, int size) {
     }
 
     @Override
-    protected InputStream onSubscribe(Subscriber<? super byte[]> subscriber) {
-        return is;
+    protected InputStream generateState() {
+        return this.is;
     }
 
     @Override
-    protected void next(SubscriptionState<byte[], InputStream> state) {
-
-        InputStream is = state.state();
+    protected InputStream next(InputStream state, Observer<? super byte[]> observer) {
         byte[] buffer = new byte[size];
         try {
-            int count = is.read(buffer);
+            int count = state.read(buffer);
             if (count == -1)
-                state.onCompleted();
+                observer.onCompleted();
             else if (count < size)
-                state.onNext(Arrays.copyOf(buffer, count));
+                observer.onNext(Arrays.copyOf(buffer, count));
             else
-                state.onNext(buffer);
+                observer.onNext(buffer);
         } catch (IOException e) {
-            state.onError(e);
+            observer.onError(e);
         }
+        return state;
     }
 }

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

@@ -1,12 +1,12 @@
 package rx.internal.operators;
 
+import rx.Observer;
+import rx.observables.SyncOnSubscribe;
+
 import java.io.IOException;
 import java.io.Reader;
 
-import rx.Subscriber;
-import rx.observables.AbstractOnSubscribe;
-
-public final class OnSubscribeReader extends AbstractOnSubscribe<String, Reader> {
+public final class OnSubscribeReader extends SyncOnSubscribe<Reader, String> {
 
     private final Reader reader;
     private final int size;
@@ -17,23 +17,22 @@ public OnSubscribeReader(Reader reader, int size) {
     }
 
     @Override
-    protected Reader onSubscribe(Subscriber<? super String> subscriber) {
-        return reader;
+    protected Reader generateState() {
+        return this.reader;
     }
 
     @Override
-    protected void next(SubscriptionState<String, Reader> state) {
-
-        Reader reader = state.state();
+    protected Reader next(Reader state, Observer<? super String> observer) {
         char[] buffer = new char[size];
         try {
-            int count = reader.read(buffer);
+            int count = state.read(buffer);
             if (count == -1)
-                state.onCompleted();
+                observer.onCompleted();
             else
-                state.onNext(String.valueOf(buffer, 0, count));
+                observer.onNext(String.valueOf(buffer, 0, count));
         } catch (IOException e) {
-            state.onError(e);
+            observer.onError(e);
         }
+        return state;
     }
 }