package au.gov.amsa.streams;

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> {

	private final InputStream is;
	private final int size;

	public OnSubscribeInputStream(InputStream is, int size) {
		this.is = is;
		this.size = size;
	}

	@Override
	protected InputStream onSubscribe(Subscriber<? super byte[]> subscriber) {
		return is;
	}

	@Override
	protected void next(
			rx.observables.AbstractOnSubscribe.SubscriptionState<byte[], InputStream> state) {

		InputStream is = state.state();
		byte[] buffer = new byte[size];
		try {
			int count = is.read(buffer);
			if (count == -1)
				state.onCompleted();
			else
				state.onNext(Arrays.copyOf(buffer, count));
		} catch (IOException e) {
			state.onError(e);
		}
	}
}

@@ -220,19 +220,17 @@
  extends
-%0A%09%09
+ 
 Abstract
@@ -265,17 +265,20 @@
 eam%3E %7B%0A%0A
-%09
+    
 private 
@@ -299,17 +299,20 @@
 eam is;%0A
-%09
+    
 private 
@@ -328,17 +328,20 @@
  size;%0A%0A
-%09
+    
 public O
@@ -390,18 +390,24 @@
 size) %7B%0A
-%09%09
+        
 this.is 
@@ -412,18 +412,24 @@
 s = is;%0A
-%09%09
+        
 this.siz
@@ -438,21 +438,27 @@
 = size;%0A
-%09%7D%0A%0A%09
+    %7D%0A%0A    
 @Overrid
@@ -451,33 +451,36 @@
 %0A%0A    @Override%0A
-%09
+    
 protected InputS
@@ -538,18 +538,24 @@
 iber) %7B%0A
-%09%09
+        
 return i
@@ -561,13 +561,19 @@
 is;%0A
-%09%7D%0A%0A%09
+    %7D%0A%0A    
 @Ove
@@ -578,17 +578,20 @@
 verride%0A
-%09
+    
 protecte
@@ -603,19 +603,28 @@
 d next(%0A
-%09%09%09
+            
 rx.obser
@@ -699,18 +699,24 @@
 ate) %7B%0A%0A
-%09%09
+        
 InputStr
@@ -739,18 +739,24 @@
 tate();%0A
-%09%09
+        
 byte%5B%5D b
@@ -783,19 +783,34 @@
 e%5D;%0A
-%09%09try %7B%0A%09%09%09
+        try %7B%0A            
 int 
@@ -834,19 +834,28 @@
 uffer);%0A
-%09%09%09
+            
 if (coun
@@ -863,20 +863,32 @@
  == -1)%0A
-%09%09%09%09
+                
 state.on
@@ -904,66 +904,166 @@
 ();%0A
-%09%09%09else%0A%09%09%09%09state.onNext(Arrays.copyOf(buffer, count));%0A%09%09
+            else if (count %3C size)%0A                state.onNext(Arrays.copyOf(buffer, count));%0A            else%0A                state.onNext(buffer);%0A        
 %7D ca
@@ -1088,11 +1088,20 @@
 ) %7B%0A
-%09%09%09
+            
 stat
@@ -1118,13 +1118,22 @@
 e);%0A
-%09%09%7D%0A%09
+        %7D%0A    
 %7D%0A%7D%0A

optimize OnSubscribeInputStream