package com.github.davidmoten.rx.web;

import java.io.ByteArrayOutputStream;

import rx.Observable.Operator;
import rx.Observer;
import rx.Subscriber;
import rx.observers.Subscribers;

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(
					final Subscriber<? super byte[]> subscriber) {
				Subscriber<byte[]> sub = Subscribers
						.from(new Observer<byte[]>() {

							@Override
							public void onCompleted() {
								subscriber.onCompleted();
							}

							@Override
							public void onError(Throwable e) {
								subscriber.onError(e);
							}

							@Override
							public void onNext(byte[] b) {
								int num = Math.min(b.length, n - buffer.size());
								if (num > 0)
									buffer.write(b, 0, num);
								if (buffer.size() == n) {
									subscriber.onNext(buffer.toByteArray());
									buffer.reset();
									subscriber.onCompleted();
								}
							}
						});
				subscriber.add(sub);
				return sub;
			}
		};

	}
}

package com.github.davidmoten.rx.web;

import java.io.ByteArrayOutputStream;

import rx.Observable.Operator;
import rx.Subscriber;
import rx.subscriptions.CompositeSubscription;

public class ByteObservable {

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

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

				final ByteArrayOutputStream buffer = new ByteArrayOutputStream();
				CompositeSubscription parent = new CompositeSubscription();
				if (n == 0) {
					child.onCompleted();
					parent.unsubscribe();
				}
				child.add(parent);
				return new Subscriber<byte[]>(parent) {

					boolean completed = false;

					@Override
					public void onCompleted() {
						if (!completed) {
							child.onCompleted();
						}
					}

					@Override
					public void onError(Throwable e) {
						if (!completed) {
							child.onError(e);
						}
					}

					@Override
					public void onNext(byte[] b) {
						if (!isUnsubscribed()) {
							synchronized (this) {
								int num = Math.min(b.length, n - buffer.size());
								if (num > 0)
									buffer.write(b, 0, num);
							}
							if (buffer.size() == n) {
								completed = true;
								child.onNext(buffer.toByteArray());
								buffer.reset();
								child.onCompleted();
								unsubscribe();
							}
						}
					}

				};
			}
		};

	}
}

add byteobs