Skip to content

Commit f49c858

Browse files
committed
Merge pull request #36 from trsc/1.x
update rx dependency to 1.1.1
2 parents d8306c3 + f7e8244 commit f49c858

File tree

3 files changed

+26
-28
lines changed

3 files changed

+26
-28
lines changed

build.gradle

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -7,7 +7,7 @@ apply plugin: 'rxjava-project'
77
apply plugin: 'java'
88

99
dependencies {
10-
compile 'io.reactivex:rxjava:1.0.12'
10+
compile 'io.reactivex:rxjava:1.1.1'
1111
testCompile 'junit:junit-dep:4.10'
1212
testCompile 'org.mockito:mockito-core:1.8.5'
1313
}
Lines changed: 13 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -1,13 +1,13 @@
11
package rx.internal.operators;
22

3+
import rx.Observer;
4+
import rx.observables.SyncOnSubscribe;
5+
36
import java.io.IOException;
47
import java.io.InputStream;
58
import java.util.Arrays;
69

7-
import rx.Subscriber;
8-
import rx.observables.AbstractOnSubscribe;
9-
10-
public final class OnSubscribeInputStream extends AbstractOnSubscribe<byte[], InputStream> {
10+
public final class OnSubscribeInputStream extends SyncOnSubscribe<InputStream, byte[]> {
1111

1212
private final InputStream is;
1313
private final int size;
@@ -18,25 +18,24 @@ public OnSubscribeInputStream(InputStream is, int size) {
1818
}
1919

2020
@Override
21-
protected InputStream onSubscribe(Subscriber<? super byte[]> subscriber) {
22-
return is;
21+
protected InputStream generateState() {
22+
return this.is;
2323
}
2424

2525
@Override
26-
protected void next(SubscriptionState<byte[], InputStream> state) {
27-
28-
InputStream is = state.state();
26+
protected InputStream next(InputStream state, Observer<? super byte[]> observer) {
2927
byte[] buffer = new byte[size];
3028
try {
31-
int count = is.read(buffer);
29+
int count = state.read(buffer);
3230
if (count == -1)
33-
state.onCompleted();
31+
observer.onCompleted();
3432
else if (count < size)
35-
state.onNext(Arrays.copyOf(buffer, count));
33+
observer.onNext(Arrays.copyOf(buffer, count));
3634
else
37-
state.onNext(buffer);
35+
observer.onNext(buffer);
3836
} catch (IOException e) {
39-
state.onError(e);
37+
observer.onError(e);
4038
}
39+
return state;
4140
}
4241
}
Lines changed: 12 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -1,12 +1,12 @@
11
package rx.internal.operators;
22

3+
import rx.Observer;
4+
import rx.observables.SyncOnSubscribe;
5+
36
import java.io.IOException;
47
import java.io.Reader;
58

6-
import rx.Subscriber;
7-
import rx.observables.AbstractOnSubscribe;
8-
9-
public final class OnSubscribeReader extends AbstractOnSubscribe<String, Reader> {
9+
public final class OnSubscribeReader extends SyncOnSubscribe<Reader, String> {
1010

1111
private final Reader reader;
1212
private final int size;
@@ -17,23 +17,22 @@ public OnSubscribeReader(Reader reader, int size) {
1717
}
1818

1919
@Override
20-
protected Reader onSubscribe(Subscriber<? super String> subscriber) {
21-
return reader;
20+
protected Reader generateState() {
21+
return this.reader;
2222
}
2323

2424
@Override
25-
protected void next(SubscriptionState<String, Reader> state) {
26-
27-
Reader reader = state.state();
25+
protected Reader next(Reader state, Observer<? super String> observer) {
2826
char[] buffer = new char[size];
2927
try {
30-
int count = reader.read(buffer);
28+
int count = state.read(buffer);
3129
if (count == -1)
32-
state.onCompleted();
30+
observer.onCompleted();
3331
else
34-
state.onNext(String.valueOf(buffer, 0, count));
32+
observer.onNext(String.valueOf(buffer, 0, count));
3533
} catch (IOException e) {
36-
state.onError(e);
34+
observer.onError(e);
3735
}
36+
return state;
3837
}
3938
}

0 commit comments

Comments
 (0)