package au.gov.amsa.streams;

import static org.junit.Assert.assertEquals;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;

import java.io.IOException;
import java.net.InetAddress;
import java.net.Socket;
import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

import org.junit.Test;

import rx.Observable;
import rx.Observable.OnSubscribe;
import rx.Subscriber;
import rx.functions.Action1;
import rx.functions.Func1;
import rx.schedulers.Schedulers;
import au.gov.amsa.util.PortFinder;

public class LinesTest {

	@Test
	public void testLinesFromARegularlyEmittingSource()
			throws InterruptedException {
		Observable<String> source = Observable.range(1, 1000000)
				.map(new Func1<Integer, String>() {
					@Override
					public String call(Integer n) {
						return n.toString();
					}
				}).delay(100, TimeUnit.MILLISECONDS, Schedulers.immediate());
		ExecutorService executor = Executors.newFixedThreadPool(2);
		int port = PortFinder.findFreePort();
		final StringServer server = StringServer.create(source, port);
		Runnable runnable = new Runnable() {
			@Override
			public void run() {
				server.start();
			}
		};
		executor.execute(runnable);
		List<String> list = Strings
				.from("localhost", port, 1000, 1000, StandardCharsets.UTF_8)
				.take(5).toList().toBlocking().single();
		System.out.println(list);
		server.stop();
		executor.awaitTermination(1000, TimeUnit.MILLISECONDS);
	}

	@Test
	public void testTrim() {
		assertEquals("trimmed", Strings.TRIM.call("  \ttrimmed\r\n   "));
	}

	@Test
	public void testTrimOnNullInputReturnsNull() {
		assertEquals(null, Strings.TRIM.call(null));
	}

	@Test(expected = RuntimeException.class)
	public void testSocketCreatorThrowsException() {
		Strings.socketCreator("non-existent-host", 1234).call();
	}

	@Test(expected = RuntimeException.class)
	public void testSocketObservableFactoryOnException() throws IOException {
		Socket socket = mock(Socket.class);
		doThrow(new IOException("hi")).when(socket).getInputStream();
		Strings.socketObservableFactory(StandardCharsets.UTF_8).call(socket);
	}

	@Test
	public void testSocketDisposerOnException() throws IOException {
		Socket socket = mock(Socket.class);
		InetAddress address = mock(InetAddress.class);
		doReturn(address).when(socket).getInetAddress();
		doReturn("ahost").when(address).getHostAddress();
		doReturn(1234).when(socket).getPort();
		doThrow(new IOException("hi")).when(socket).close();
		Strings.socketDisposer().call(socket);
	}

	private static class MySource implements OnSubscribe<String> {

		private final String value;

		public MySource(String value) {
			this.value = value;
		}

		@Override
		public void call(Subscriber<? super String> sub) {
			while (!sub.isUnsubscribed()) {
				try {
					Thread.sleep(30);
				} catch (InterruptedException e) {
					// do nothing
				}
				sub.onNext(value);
			}
		}

	}

	public static void main(String[] args) {
		Observable.range(1, 1000).mergeWith(Observable.range(1001, 1000))
				.subscribe(new Action1<Integer>() {
					@Override
					public void call(Integer n) {
						System.out.println(n);
					}
				});
		System.exit(0);
		Observable.create(new MySource("a"))
				.subscribeOn(Schedulers.newThread())
				.mergeWith(Observable.create(new MySource("b")))
				.subscribe(new Action1<String>() {

					@Override
					public void call(String s) {
						if (!s.equals("a"))
							System.out.println(s);
					}
				});
	}
}

@@ -2963,17 +2963,18 @@
 d.sleep(
-3
+50
 0);%0A%09%09%09%09
@@ -3114,229 +3114,37 @@
 gs) 
-%7B%0A%09%09Observable.range(1, 1000).mergeWith(Observable.range(1001, 1000))%0A%09%09%09%09.subscribe(new Action1%3CInteger%3E() %7B%0A%09%09%09%09%09@Override%0A%09%09%09%09%09public void call(Integer n) %7B%0A%09%09%09%09%09%09System.out.println(n);%0A%09%09%09%09%09%7D%0A%09%09%09%09%7D);%0A%09%09System.exit(0);
+throws InterruptedException %7B
 %0A%09%09O
@@ -3144,32 +3144,37 @@
 n %7B%0A%09%09Observable
+%0A%09%09%09%09
 .create(new MySo
@@ -3216,17 +3216,17 @@
 ers.
-newThread
+immediate
 ())%0A
@@ -3240,16 +3240,23 @@
 rgeWith(
+%0A%09%09%09%09%09%09
 Observab
@@ -3283,16 +3283,61 @@
 ce(%22b%22))
+.subscribeOn(%0A%09%09%09%09%09%09%09%09Schedulers.newThread())
 )%0A%09%09%09%09.s
@@ -3369,17 +3369,16 @@
 ng%3E() %7B%0A
-%0A
 %09%09%09%09%09@Ov
@@ -3422,35 +3422,8 @@
 ) %7B%0A
-%09%09%09%09%09%09if (!s.equals(%22a%22))%0A%09
 %09%09%09%09

update main