Merge pull request #8 from davidmoten/unsub-check

add check for unsubscribed before next read

Author: akarnokd

src/main/java/rx/observables/StringObservable.java

@@ -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);

src/test/java/rx/observables/StringObservableTest.java

@@ -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;
@@ -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() {