package com.github.davidmoten.rx.web;

import java.io.ByteArrayOutputStream;

import rx.Observable.Operator;
import rx.Subscriber;

public class ByteObservable {

	public static Operator<byte[], byte[]> first(final int n) {
		return new Operator<byte[], byte[]>() {

			private final ByteArrayOutputStream buffer = new ByteArrayOutputStream();

			@Override
			public Subscriber<? super byte[]> call(
					Subscriber<? super byte[]> subscriber) {

			}
		};

	}
}

@@ -112,26 +112,79 @@
 port rx.
-Subscriber
+Observer;%0Aimport rx.Subscriber;%0Aimport rx.observers.Subscribers
 ;%0A%0Apubli
@@ -452,16 +452,22 @@
 l(%0A%09%09%09%09%09
+final 
 Subscrib
@@ -495,24 +495,679 @@
 bscriber) %7B%0A
+%09%09%09%09Subscriber%3Cbyte%5B%5D%3E sub = Subscribers%0A%09%09%09%09%09%09.from(new Observer%3Cbyte%5B%5D%3E() %7B%0A%0A%09%09%09%09%09%09%09@Override%0A%09%09%09%09%09%09%09public void onCompleted() %7B%0A%09%09%09%09%09%09%09%09subscriber.onCompleted();%0A%09%09%09%09%09%09%09%7D%0A%0A%09%09%09%09%09%09%09@Override%0A%09%09%09%09%09%09%09public void onError(Throwable e) %7B%0A%09%09%09%09%09%09%09%09subscriber.onError(e);%0A%09%09%09%09%09%09%09%7D%0A%0A%09%09%09%09%09%09%09@Override%0A%09%09%09%09%09%09%09public void onNext(byte%5B%5D b) %7B%0A%09%09%09%09%09%09%09%09int num = Math.min(b.length, n - buffer.size());%0A%09%09%09%09%09%09%09%09if (num %3E 0)%0A%09%09%09%09%09%09%09%09%09buffer.write(b, 0, num);%0A%09%09%09%09%09%09%09%09if (buffer.size() == n) %7B%0A%09%09%09%09%09%09%09%09%09subscriber.onNext(buffer.toByteArray());%0A%09%09%09%09%09%09%09%09%09buffer.reset();%0A%09%09%09%09%09%09%09%09%09subscriber.onCompleted();%0A%09%09%09%09%09%09%09%09%7D%0A%09%09%09%09%09%09%09%7D%0A%09%09%09%09%09%09%7D);%0A%09%09%09%09subscriber.add(sub);%0A%09%09%09%09return sub;
 %0A%09%09%09%7D%0A%09%09%7D;%0A%0A

add byteobs