package se.l4.silo.engine.internal;

import java.nio.file.Path;
import java.util.Optional;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

import com.carrotsearch.randomizedtesting.RandomizedTest;
import com.carrotsearch.randomizedtesting.annotations.ThreadLeakLingering;

import org.junit.Test;

import net.jodah.concurrentunit.Waiter;
import se.l4.silo.FetchResult;
import se.l4.silo.Silo;
import se.l4.silo.binary.BinaryEntity;
import se.l4.silo.binary.BinaryEntry;
import se.l4.silo.engine.Index;
import se.l4.silo.engine.LocalSilo;
import se.l4.silo.structured.ObjectEntity;

public class ThreadSafetyTest
	extends RandomizedTest
{
	@Test
	@ThreadLeakLingering(linger = 1000)
	public void testSeveralThreadsWithBinaryEntity()
		throws Exception
	{
		Path tmp = newTempDir();
		Silo silo = LocalSilo.open(tmp)
			.addEntity("test").asBinary().done()
			.build();

		closeAfterTest(silo);

		ExecutorService executor = Executors.newFixedThreadPool(randomIntBetween(4, 8));
		Waiter waiter = new Waiter();

		BinaryEntity entity = silo.binary("test");

		int entries = scaledRandomIntBetween(1000, 10000) * 2;
		for(int i=0, n=entries; i<n; i++)
		{
			executor.submit(() -> {
				try
				{
					String id = randomUnicodeOfLengthBetween(2, 10);
					int size = randomIntBetween(100, 8192);
					entity.store(id, DataUtils.generate(size));

					executor.submit(() -> {
						try
						{
							// Try to read back the data and verify that it was stored correctly

							FetchResult<BinaryEntry> fr = entity.get(id);
							DataUtils.assertBytesEquals(fr.iterator().next().getData(), DataUtils.generate(size));
						}
						catch(Throwable t)
						{
							waiter.rethrow(t);
						}
						waiter.resume();
					});
				}
				catch(Throwable t)
				{
					waiter.rethrow(t);
				}
				waiter.resume();
			});
		}

		waiter.await(1, TimeUnit.MINUTES, entries);

		executor.shutdown();
		executor.awaitTermination(1, TimeUnit.MINUTES);
	}

	@Test
	@ThreadLeakLingering(linger = 1000)
	public void testSeveralThreadsWithObjectEntity()
		throws Exception
	{
		Path tmp = newTempDir();
		Silo silo = LocalSilo.open(tmp)
			.addEntity("test")
				.asStructured()
				.defineField("name", "string")
				.add("byName", Index::queryEngine)
					.addField("name")
					.done()
				.done()
			.build();

		closeAfterTest(silo);

		ExecutorService executor = Executors.newFixedThreadPool(randomIntBetween(4, 8));
		Waiter waiter = new Waiter();

		ObjectEntity<TestUserData> entity = silo.structured("test")
			.asObject(TestUserData.class, TestUserData::getId);

		int entries = scaledRandomIntBetween(1000, 10000) * 2;
		for(int i=0, n=entries; i<n; i++)
		{
			executor.submit(() -> {
				try
				{
					int id = randomIntBetween(100, 1000000);
					String name = randomRealisticUnicodeOfLengthBetween(4, 10);
					entity.store(new TestUserData(id, name, 1, false));

					executor.submit(() -> {
						try
						{
							// Try to read back the data and verify that it was stored correctly

							Optional<TestUserData> td = entity.get(id);
							waiter.assertEquals(new TestUserData(id, name, 1, false), td.get());
						}
						catch(Throwable t)
						{
							waiter.rethrow(t);
						}
						waiter.resume();
					});
				}
				catch(Throwable t)
				{
					waiter.rethrow(t);
				}
				waiter.resume();
			});
		}

		waiter.await(1, TimeUnit.MINUTES, entries);

		executor.shutdown();
		executor.awaitTermination(1, TimeUnit.MINUTES);
	}
}

@@ -339,16 +339,79 @@
 ngering;
+%0Aimport com.google.common.util.concurrent.ThreadFactoryBuilder;
 %0A%0Aimport
@@ -802,32 +802,33 @@
 ng(linger = 1000
+0
 )%0A%09public void t
@@ -1075,32 +1075,36 @@
 FixedThreadPool(
+%0A%09%09%09
 randomIntBetween
@@ -1101,32 +1101,126 @@
 IntBetween(4, 8)
+,%0A%09%09%09new ThreadFactoryBuilder()%0A%09%09%09%09.setNameFormat(%22binary-entity-threads-%25d%22)%0A%09%09%09%09.build()%0A%09%09
 );%0A%09%09Waiter wait
@@ -2214,16 +2214,17 @@
 r = 1000
+0
 )%0A%09publi
@@ -2611,16 +2611,20 @@
 eadPool(
+%0A%09%09%09
 randomIn
@@ -2637,16 +2637,110 @@
 en(4, 8)
+,%0A%09%09%09new ThreadFactoryBuilder()%0A%09%09%09%09.setNameFormat(%22object-entity-threads-%25d%22)%0A%09%09%09%09.build()%0A%09%09
 );%0A%09%09Wai

Increase linger time for ThreadSafetyTest