/*
 * Copyright (c) 2011-2016 Pivotal Software Inc, All Rights Reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *       http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package reactor.aeron;

import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import reactor.aeron.publisher.AeronPublisher;
import reactor.aeron.subscriber.AeronSubscriber;
import reactor.aeron.support.AeronTestUtils;
import reactor.aeron.support.ThreadSnapshot;
import reactor.core.subscriber.test.DataTestSubscriber;
import reactor.io.buffer.Buffer;
import reactor.io.net.tcp.support.SocketUtils;
import reactor.rx.Streams;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;

import static org.junit.Assert.assertTrue;

/**
 * @author Anatoly Kadyshev
 */
public abstract class CommonSubscriberPublisherTest {

	public static final int TIMEOUT_SECS = 5;

	final int senderPort = SocketUtils.findAvailableTcpPort();

	private ThreadSnapshot threadSnapshot;

	@Before
	public void doSetup() {
		threadSnapshot = new ThreadSnapshot().take();

		AeronTestUtils.setAeronEnvProps();
	}

	@After
	public void doTearDown() throws InterruptedException {
		assertTrue(threadSnapshot.takeAndCompare(new String[] {"hash", "global"},
				TimeUnit.SECONDS.toMillis(TIMEOUT_SECS)));
	}

	protected abstract Context createContext();

	protected List<Buffer> createBuffers(int n) {
		List<Buffer> items = new ArrayList<>(n);
		for (int i = 1; i <= n; i++) {
			items.add(Buffer.wrap("" + i));
		}
		return items;
	}

	@Test
	public void testNextSignalIsReceivedByPublisher() throws InterruptedException {
		AeronSubscriber subscriber = AeronSubscriber.create(createContext());

		Streams.just(
				Buffer.wrap("One"),
				Buffer.wrap("Two"),
				Buffer.wrap("Three"))
				.subscribe(subscriber);

		AeronPublisher publisher = AeronPublisher.create(createContext());

		DataTestSubscriber<String>clientSubscriber = DataTestSubscriber.createWithTimeoutSecs(TIMEOUT_SECS);
		publisher.subscribe(clientSubscriber);


		clientSubscriber.requestUnboundedWithTimeout();


		clientSubscriber.assertNextSignals("One", "Two", "Three");
		clientSubscriber.assertCompleteReceived();
	}

	@Test
	public void testErrorShutsDownAeronPublisherAndSubscriber() throws InterruptedException {
		AeronSubscriber subscriber = AeronSubscriber.create(createContext());
		AeronPublisher publisher = AeronPublisher.create(createContext());

		DataTestSubscriber<String>clientSubscriber = DataTestSubscriber.createWithTimeoutSecs(TIMEOUT_SECS);
		publisher.subscribe(clientSubscriber);

		clientSubscriber.requestUnboundedWithTimeout();

		Thread.sleep(1000);

		Streams.<Buffer, Throwable>fail(new RuntimeException("Something went wrong"))
				.subscribe(subscriber);

		clientSubscriber.assertErrorReceived();
	}

}

@@ -630,16 +630,17 @@
 se.%0A */%0A
+%0A
 package 
@@ -655,16 +655,106 @@
 aeron;%0A%0A
+import java.util.ArrayList;%0Aimport java.util.List;%0Aimport java.util.concurrent.TimeUnit;%0A%0A
 import o
@@ -1051,32 +1051,54 @@
 TestSubscriber;%0A
+import reactor.io.IO;%0A
 import reactor.i
@@ -1193,98 +1193,8 @@
 s;%0A%0A
-import java.util.ArrayList;%0Aimport java.util.List;%0Aimport java.util.concurrent.TimeUnit;%0A%0A
 impo
@@ -1713,17 +1713,16 @@
 String%5B%5D
- 
 %7B%22hash%22,
@@ -2188,21 +2188,16 @@
 ms.just(
-%0A%09%09%09%09
 Buffer.w
@@ -2207,21 +2207,17 @@
 (%22One%22),
-%0A%09%09%09%09
+ 
 Buffer.w
@@ -2227,21 +2227,17 @@
 (%22Two%22),
-%0A%09%09%09%09
+ 
 Buffer.w
@@ -2248,26 +2248,31 @@
 %22Three%22))%0A%09%09
-%09%09
+       
 .subscribe(s
@@ -2374,32 +2374,33 @@
 bscriber%3CString%3E
+ 
 clientSubscriber
@@ -2452,32 +2452,50 @@
 IMEOUT_SECS);%0A%09%09
+IO.bufferToString(
 publisher.subscr
@@ -2479,32 +2479,33 @@
 String(publisher
+)
 .subscribe(clien
@@ -2515,25 +2515,24 @@
 bscriber);%0A%0A
-%0A
 %09%09clientSubs
@@ -2570,17 +2570,16 @@
 out();%0A%0A
-%0A
 %09%09client
@@ -2948,16 +2948,17 @@
 %3CString%3E
+ 
 clientSu
@@ -3026,16 +3026,34 @@
 ECS);%0A%09%09
+IO.bufferToString(
 publishe
@@ -3053,16 +3053,17 @@
 ublisher
+)
 .subscri
@@ -3238,21 +3238,16 @@
 wrong%22))
-%0A%09%09%09%09
 .subscri

fix compile issue