content: package io.activej.launchers.crdt;
  
  import io.activej.config.Config;
  import io.activej.crdt.CrdtData;
  import io.activej.crdt.RemoteCrdtStorage;
  import io.activej.crdt.storage.ICrdtStorage;
  import io.activej.crdt.util.CrdtDataBinarySerializer;
  import io.activej.datastream.consumer.StreamConsumers;
  import io.activej.datastream.supplier.StreamSuppliers;
  import io.activej.dns.DnsClient;
  import io.activej.fs.FileSystem;
  import io.activej.fs.IFileSystem;
  import io.activej.http.HttpClient;
  import io.activej.http.HttpRequest;
  import io.activej.http.IHttpClient;
  import io.activej.inject.annotation.Provides;
  import io.activej.inject.module.AbstractModule;
  import io.activej.inject.module.Module;
  import io.activej.promise.Promises;
  import io.activej.promise.jmx.PromiseStats;
  import io.activej.reactor.Reactor;
  import io.activej.reactor.nio.NioReactor;
  import io.activej.test.rules.ActivePromisesRule;
  import io.activej.test.rules.ByteBufRule;
  import io.activej.test.rules.EventloopRule;
  import org.junit.ClassRule;
  import org.junit.Ignore;
  import org.junit.Rule;
  import org.junit.Test;
  
  import java.net.InetSocketAddress;
  import java.time.Duration;
  import java.util.concurrent.Executor;
  import java.util.stream.IntStream;
  
  import static io.activej.config.converter.ConfigConverters.ofExecutor;
  import static io.activej.config.converter.ConfigConverters.ofPath;
  import static io.activej.crdt.function.CrdtFunction.ignoringTimestamp;
  import static io.activej.crdt.json.JsonCodecs.ofCrdtData;
  import static io.activej.http.HttpMethod.PUT;
  import static io.activej.http.HttpUtils.inetAddress;
  import static io.activej.json.JsonCodecs.ofInteger;
  import static io.activej.json.JsonCodecs.ofString;
  import static io.activej.json.JsonUtils.toJsonBytes;
  import static io.activej.promise.TestUtils.await;
  import static io.activej.serializer.BinarySerializers.INT_SERIALIZER;
  import static io.activej.serializer.BinarySerializers.UTF8_SERIALIZER;
  import static io.activej.test.TestUtils.assertCompleteFn;
  
  @Ignore("manual demos")
  public final class CrdtClusterTest {
  	@ClassRule
  	public static final EventloopRule eventloopRule = new EventloopRule();
  
  	@ClassRule
  	public static final ByteBufRule byteBufRule = new ByteBufRule();
  
  	@Rule
  	public final ActivePromisesRule activePromisesRule = new ActivePromisesRule();
  
  	static class TestNodeLauncher extends CrdtNodeLauncher<String, Integer> {
  		private final Config config;
  
  		public TestNodeLauncher(Config config) {
  			this.config = config;
  		}
  
  		@Override
  		protected Module getOverrideModule() {
  			return new AbstractModule() {
  				@Provides
  				Config config() {
  					return config;
  				}
  			};
  		}
  
  		@Override
  		protected CrdtNodeLogicModule<String, Integer> getBusinessLogicModule() {
  			return new CrdtNodeLogicModule<String, Integer>() {
  				@Override
  				protected void configure() {
  					install(new CrdtHttpModule<String, Integer>() {});
  				}
  
  				@Provides
  				CrdtDescriptor<String, Integer> descriptor() {
  					return new CrdtDescriptor<>(
  						ignoringTimestamp(Integer::max),
  						new CrdtDataBinarySerializer<>(UTF8_SERIALIZER, INT_SERIALIZER),
  						ofString(),
  						ofInteger());
  				}
  
  				@Provides
  				Executor executor(Config config) {
  					return config.get(ofExecutor(), "crdt.local.executor");
  				}
  
  				@Provides
  				IFileSystem fileSystem(Reactor reactor, Executor executor, Config config) {
  					return FileSystem.create(reactor, executor, config.get(ofPath(), "crdt.local.path"));
  				}
  			};
  		}
  	}
  
  	@Test
  	public void startFirst() throws Exception {
  		new TestNodeLauncher(Config.create()
  			.with("crdt.http.listenAddresses", "localhost:7000")
  			.with("crdt.server.listenAddresses", "localhost:8000")
  			.with("crdt.cluster.server.listenAddresses", "localhost:9000")
  			.with("crdt.local.path", "/tmp/TESTS/crdt")
  			.with("crdt.cluster.localPartitionId", "first")
  			.with("crdt.cluster.replicationCount", "2")
  			.with("crdt.cluster.partitions.second", "localhost:8001")
  			//				.with("crdt.cluster.partitions.file", "localhost:8002")
  		).launch(new String[0]);
  	}
  
  	@Test
  	public void startSecond() throws Exception {
  		new TestNodeLauncher(Config.create()
  			.with("crdt.http.listenAddresses", "localhost:7001")
  			.with("crdt.server.listenAddresses", "localhost:8001")
  			.with("crdt.cluster.server.listenAddresses", "localhost:9001")
  			.with("crdt.local.path", "/tmp/TESTS/crdt")
  			.with("crdt.cluster.localPartitionId", "second")
  			.with("crdt.cluster.replicationCount", "2")
  			.with("crdt.cluster.partitions.first", "localhost:8000")
  			//				.with("crdt.cluster.partitions.file", "localhost:8002")
  		).launch(new String[0]);
  	}
  
  	@Test
  	public void startFileServer() throws Exception {
  		new CrdtFileServerLauncher<String, Integer>() {
  			@Override
  			protected CrdtFileServerLogicModule<String, Integer> getBusinessLogicModule() {
  				return new CrdtFileServerLogicModule<>() {};
  			}
  
  			@Override
  			protected Module getOverrideModule() {
  				return new AbstractModule() {
  					@Provides
  					Config config() {
  						return Config.create()
  							.with("crdt.localPath", "/tmp/TESTS/fileServer")
  							.with("crdt.server.listenAddresses", "localhost:8002");
  					}
  				};
  			}
  
  			@Provides
  			CrdtDescriptor<String, Integer> descriptor() {
  				return new CrdtDescriptor<>(
  					ignoringTimestamp(Integer::max),
  					new CrdtDataBinarySerializer<>(UTF8_SERIALIZER, INT_SERIALIZER),
  					ofString(),
  					ofInteger());
  			}
  		}.launch(new String[0]);
  	}
  
  	@Test
  	public void uploadWithHTTP() {
  		NioReactor reactor = Reactor.getCurrentReactor();
  		DnsClient dnsClient = DnsClient.create(reactor, inetAddress("8.8.8.8"));
  		IHttpClient client = HttpClient.create(reactor, dnsClient);
  
  		PromiseStats uploadStat = PromiseStats.create(Duration.ofSeconds(5));
  
  		Promises.sequence(IntStream.range(0, 1_000_000)
  				.mapToObj(i ->
  					() -> client.request(HttpRequest.builder(PUT, "http://127.0.0.1:7000")
  							.withBody(toJsonBytes(
  								ofCrdtData(ofString(), ofInteger()),
  								new CrdtData<>(
  									"value_" + i,
  									reactor.currentTimeMillis(),
  									i
  								)))
  							.build())
  						.toVoid()))
  			.whenException(Exception::printStackTrace)
  			.whenComplete(uploadStat.recordStats())
  			.whenComplete(assertCompleteFn($ -> System.out.println(uploadStat)));
  
  		// RemoteCrdtClient<String, Integer> client = RemoteCrdtClient.create(eventloop, ADDRESS, CRDT_DATA_SERIALIZER);
  		//
  		// StreamProducer.ofStream(IntStream.range(0, 1000000)
  		// 		.mapToObj(i -> new CrdtData<>("key_" + i, i))).streamTo(client.uploadStream())
  		// 		.getEndOfStream()
  		// 		.thenRun(() -> System.out.println("finished"));
  	}
  
  	@Test
  	public void uploadWithStreams() {
  		ICrdtStorage<String, Integer> client = RemoteCrdtStorage.create(Reactor.getCurrentReactor(), new InetSocketAddress("localhost", 9000), UTF8_SERIALIZER, INT_SERIALIZER);
  
  		PromiseStats uploadStat = PromiseStats.create(Duration.ofSeconds(5));
  
  		StreamSuppliers.ofStream(IntStream.range(0, 1000000)
  				.mapToObj(i -> new CrdtData<>("value_" + i, Reactor.getCurrentReactor().currentTimeMillis(), i)))
  			.streamTo(StreamConsumers.ofPromise(client.upload()))
  			.whenComplete(uploadStat.recordStats())
  			.whenComplete(assertCompleteFn($ -> {
  				System.out.println(uploadStat);
  				System.out.println("finished");
  			}));
  	}
  
  	@Test
  	public void downloadStuff() {
  		ICrdtStorage<String, Integer> client = RemoteCrdtStorage.create(Reactor.getCurrentReactor(), new InetSocketAddress(9001), UTF8_SERIALIZER, INT_SERIALIZER);
  
  		await(client.download().then(supplier -> supplier.streamTo(StreamConsumers.ofConsumer(System.out::println))));
  	}
  }
  