package io.activej.rpc;

import io.activej.common.time.Stopwatch;
import io.activej.eventloop.Eventloop;
import io.activej.promise.Promise;
import io.activej.reactor.AbstractNioReactive;
import io.activej.reactor.Reactor;
import io.activej.reactor.nio.NioReactor;
import io.activej.rpc.client.RpcClient;
import io.activej.rpc.protocol.RpcRemoteException;
import io.activej.rpc.server.RpcRequestHandler;
import io.activej.rpc.server.RpcServer;
import io.activej.serializer.annotations.SerializeRecord;
import io.activej.test.rules.ActivePromisesRule;
import io.activej.test.rules.ByteBufRule;
import io.activej.test.rules.ClassBuilderConstantsRule;
import io.activej.test.rules.EventloopRule;
import org.junit.*;

import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.atomic.AtomicInteger;

import static io.activej.rpc.client.sender.strategy.RpcStrategies.server;
import static io.activej.test.TestUtils.assertCompleteFn;
import static io.activej.test.TestUtils.getFreePort;
import static java.util.concurrent.TimeUnit.MICROSECONDS;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.fail;

public final class RpcHelloWorldTest {

	@ClassRule
	public static final ByteBufRule byteBufRule = new ByteBufRule();

	@Rule
	public final EventloopRule eventloopRule = new EventloopRule();

	@Rule
	public final ActivePromisesRule activePromisesRule = new ActivePromisesRule();

	@Rule
	public final ClassBuilderConstantsRule classBuilderConstantsRule = new ClassBuilderConstantsRule();

	private interface HelloService {
		String hello(String name) throws Exception;
	}

	@SerializeRecord
	public record HelloRequest(String name) {}

	@SerializeRecord
	public record HelloResponse(String message) {}

	private static RpcRequestHandler<HelloRequest, HelloResponse> helloServiceRequestHandler(HelloService helloService) {
		return request -> {
			String result;
			try {
				result = helloService.hello(request.name);
			} catch (Exception e) {
				return Promise.ofException((Exception) e);
			}
			return Promise.of(new HelloResponse(result));
		};
	}

	private static RpcServer createServer(NioReactor reactor) {
		return RpcServer.builder(reactor)
			.withMessageTypes(HelloRequest.class, HelloResponse.class)
			.withHandler(HelloRequest.class, helloServiceRequestHandler(name -> {
				if (name.equals("--")) {
					throw new Exception("Illegal name");
				}
				return "Hello, " + name + "!";
			}))
			.withListenPort(port)
			.build();
	}

	private static class BlockingHelloClient extends AbstractNioReactive implements HelloService, AutoCloseable {
		private final RpcClient rpcClient;

		public BlockingHelloClient(NioReactor reactor) throws Exception {
			super(reactor);
			this.rpcClient = RpcClient.builder(reactor)
				.withMessageTypes(HelloRequest.class, HelloResponse.class)
				.withStrategy(server(new InetSocketAddress(InetAddress.getByName("127.0.0.1"), port)))
				.build();

			rpcClient.startFuture().get();
		}

		@Override
		public String hello(String name) throws Exception {
			try {
				return rpcClient.getReactor().submit(
						() -> rpcClient
							.<HelloRequest, HelloResponse>sendRequest(new HelloRequest(name), TIMEOUT))
					.get()
					.message;
			} catch (ExecutionException e) {
				//noinspection ThrowInsideCatchBlockWhichIgnoresCaughtException - cause is rethrown
				throw (Exception) e.getCause();
			}
		}

		@Override
		public void close() throws Exception {
			rpcClient.stopFuture().get();
		}
	}

	private static final int TIMEOUT = 1500;
	private static int port;
	private RpcServer server;

	@Before
	public void setUp() throws Exception {
		Eventloop eventloop = Reactor.getCurrentReactor();
		port = getFreePort();
		server = createServer(eventloop);
		server.listen();
		new Thread(eventloop).start();
	}

	@Test
	public void testBlockingCall() throws Exception {
		try (BlockingHelloClient client = new BlockingHelloClient(Reactor.getCurrentReactor())) {
			for (int i = 0; i < 100; i++) {
				assertEquals("Hello, World!", client.hello("World"));
			}
		} finally {
			server.closeFuture().get();
		}
	}

	@Test
	public void testAsyncCall() throws Exception {
		int requestCount = 10;

		try (BlockingHelloClient client = new BlockingHelloClient(Reactor.getCurrentReactor())) {
			CountDownLatch latch = new CountDownLatch(requestCount);
			for (int i = 0; i < requestCount; i++) {
				String name = "World" + i;
				client.getReactor().execute(() -> client.rpcClient.<HelloRequest, HelloResponse>sendRequest(new HelloRequest(name), TIMEOUT)
					.whenComplete(latch::countDown)
					.whenComplete(assertCompleteFn(response -> assertEquals("Hello, " + name + "!", response.message))));
			}
			latch.await();
		} finally {
			server.closeFuture().get();
		}
	}

	@Test
	public void testBlocking2Clients() throws Exception {
		try (
			BlockingHelloClient client1 = new BlockingHelloClient(Reactor.getCurrentReactor());
			BlockingHelloClient client2 = new BlockingHelloClient(Reactor.getCurrentReactor())
		) {
			assertEquals("Hello, John!", client2.hello("John"));
			assertEquals("Hello, World!", client1.hello("World"));
		} finally {
			server.closeFuture().get();
		}
	}

	@Test
	public void testBlockingRpcException() throws Exception {
		try (BlockingHelloClient client = new BlockingHelloClient(Reactor.getCurrentReactor())) {
			client.hello("--");
			fail("Exception expected");
		} catch (RpcRemoteException e) {
			assertEquals("java.lang.Exception: Illegal name", e.getMessage());
		} finally {
			server.closeFuture().get();
		}
	}

	@Test
	public void testAsync2Clients() throws Exception {
		int requestCount = 10;

		try (
			BlockingHelloClient client1 = new BlockingHelloClient(Reactor.getCurrentReactor());
			BlockingHelloClient client2 = new BlockingHelloClient(Reactor.getCurrentReactor())
		) {
			CountDownLatch latch = new CountDownLatch(2 * requestCount);

			for (int i = 0; i < requestCount; i++) {
				String name = "world" + i;
				client1.getReactor().execute(() ->
					client1.rpcClient.<HelloRequest, HelloResponse>sendRequest(new HelloRequest(name), TIMEOUT)
						.whenComplete(latch::countDown)
						.whenComplete(assertCompleteFn(response -> assertEquals("Hello, " + name + "!", response.message))));
				client2.getReactor().execute(() ->
					client2.rpcClient.<HelloRequest, HelloResponse>sendRequest(new HelloRequest(name), TIMEOUT)
						.whenComplete(latch::countDown)
						.whenComplete(assertCompleteFn(response -> assertEquals("Hello, " + name + "!", response.message))));
			}
			latch.await();
		} finally {
			server.closeFuture().get();
		}
	}

	@Test
	@Ignore("this is not a test but a benchmark, takes a lot of time")
	public void testRejectedRequests() throws Exception {
		int count = 1_000_000;

		try (BlockingHelloClient client = new BlockingHelloClient(Reactor.getCurrentReactor())) {
			for (int t = 0; t < 10; t++) {
				AtomicInteger success = new AtomicInteger(0);
				AtomicInteger error = new AtomicInteger(0);
				CountDownLatch latch = new CountDownLatch(count);
				Stopwatch stopwatch = Stopwatch.createStarted();
				for (int i = 0; i < count; i++) {
					client.getReactor().execute(() ->
						client.rpcClient.<HelloRequest, HelloResponse>sendRequest(new HelloRequest("benchmark"), TIMEOUT)
							.whenComplete(($, e) -> {
								latch.countDown();
								(e == null ? success : error).incrementAndGet();
							}));
				}
				latch.await();
				System.out.printf("%2d: Elapsed %8s rps: %18s (%d/%d [%d])%n",
					t + 1, stopwatch.stop(), count * 1000000.0 / stopwatch.elapsed(MICROSECONDS), success.get(), count, error.get());
			}
		} finally {
			server.closeFuture().get();
		}
	}
}

