Skip to content

add check for unsubscribed before next read #8

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
akarnokd merged 1 commit into from
Jan 30, 2015
Merged

add check for unsubscribed before next read #8

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
add check for unsubscribed before next read to StringObservable.from(… 
…Reader) and unit test
  • Loading branch information
@davidmoten
davidmoten committed Dec 5, 2014
commit 9d7159e3f86d161c205e5d3e6c8c646f695e1b3a
3 changes: 2 additions & 1 deletion src/main/java/rx/observables/StringObservable.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -176,7 +176,8 @@ public void call(Subscriber<? super String> o) {
n = i.read(buffer);
while (n != -1 && !o.isUnsubscribed()) {
o.onNext(new String(buffer, 0, n));
n = i.read(buffer);
if (!o.isUnsubscribed())
n = i.read(buffer);
}
} catch (IOException e) {
o.onError(e);
Expand Down
18 changes: 18 additions & 0 deletions src/test/java/rx/observables/StringObservableTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -38,6 +38,7 @@
import java.io.ByteArrayInputStream;
import java.io.FilterReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.Reader;
import java.io.StringReader;
import java.nio.charset.Charset;
Expand Down Expand Up @@ -303,6 +304,23 @@ public void testFromReader() {
assertNotSame(inStr, outStr);
assertEquals(inStr, outStr);
}

@Test
public void testFromReaderWillUnsubscribeBeforeCallingNextRead() {
final byte[] inBytes = "test".getBytes();
final AtomicInteger numReads = new AtomicInteger(0);
ByteArrayInputStream is = new ByteArrayInputStream(inBytes) {

@Override
public synchronized int read(byte[] b, int off, int len) {
numReads.incrementAndGet();
return super.read(b, off, len);
}
};
StringObservable.from(new InputStreamReader(is)).first().toBlocking()
.single();
assertEquals(1, numReads.get());
}

@Test
public void testByLine() {
Expand Down