content: package io.activej.csp.queue;
  
  import io.activej.bytebuf.ByteBuf;
  import io.activej.common.ref.Ref;
  import io.activej.common.ref.RefInt;
  import io.activej.promise.Promise;
  import io.activej.test.rules.ByteBufRule;
  import io.activej.test.rules.EventloopRule;
  import org.junit.*;
  import org.junit.rules.TemporaryFolder;
  
  import java.io.IOException;
  import java.util.concurrent.ExecutorService;
  import java.util.concurrent.Executors;
  
  import static io.activej.bytebuf.ByteBufStrings.wrapAscii;
  import static io.activej.promise.TestUtils.await;
  import static java.nio.charset.StandardCharsets.US_ASCII;
  import static org.junit.Assert.*;
  
  public final class ChannelBufferWithFallbackTest {
  
  	@ClassRule
  	public static ByteBufRule byteBufRule = new ByteBufRule();
  
  	@ClassRule
  	public static EventloopRule eventloopRule = new EventloopRule();
  
  	@Rule
  	public final TemporaryFolder temporaryFolder = new TemporaryFolder();
  
  	private ExecutorService executor;
  
  	@Before
  	public void setUp() {
  		executor = Executors.newSingleThreadExecutor();
  	}
  
  	@After
  	public void tearDown() {
  		executor.shutdownNow();
  	}
  
  	private ChannelBufferWithFallback<ByteBuf> createBufferedQueue(RefInt secondaryCounter) {
  		return new ChannelBufferWithFallback<>(
  			new ChannelZeroBuffer<>(),
  			() -> {
  				secondaryCounter.value++;
  				try {
  					return ChannelFileBuffer.create(executor, temporaryFolder.newFile().toPath());
  				} catch (IOException e) {
  					throw new AssertionError(e);
  				}
  			});
  	}
  
  	@Test
  	public void bufferedQueue() {
  		RefInt counter = new RefInt(0);
  		ChannelBufferWithFallback<ByteBuf> q = createBufferedQueue(counter);
  
  		assertTrue(q.isExhausted());
  		assertFalse(q.isSaturated());
  
  		await(q.put(wrapAscii("hello world, this is a byte buffer\n")));
  		await(q.put(wrapAscii("second: hello world, this is a byte buffer")));
  
  		assertFalse(q.isExhausted());
  		assertFalse(q.isSaturated());
  
  		String taken = await(q.take()).asString(US_ASCII);
  
  		assertTrue(q.isExhausted());
  		assertFalse(q.isSaturated());
  
  		await(q.put(wrapAscii("another input\n")));
  		await(q.put(wrapAscii("second buffer")));
  		await(q.put(wrapAscii(" and a third one")));
  
  		String taken2 = await(q.take()).asString(US_ASCII);
  
  		assertEquals("hello world, this is a byte buffer\nsecond: hello world, this is a byte buffer", taken);
  		assertEquals("another input\nsecond buffer and a third one", taken2);
  
  		// should be 1 and not 2, because we've exhausted the secondary buffer,
  		// but then we are putting items in - primary queue were full and still is full at this point
  		assertEquals(1, counter.value);
  
  		q.close();
  	}
  
  	@Test
  	public void bufferedQueueTakeWait() {
  		RefInt counter = new RefInt(0);
  		ChannelBufferWithFallback<ByteBuf> cfq = createBufferedQueue(counter);
  
  		Ref<ByteBuf> ref = new Ref<>();
  
  		cfq.take().whenResult(ref::set);
  
  		Promise<Void> fastPut = cfq.put(wrapAscii("this is a string for testing"));
  		assertTrue(fastPut.isComplete());
  
  		ByteBuf res = ref.get();
  		assertNotNull(res);
  		assertEquals("this is a string for testing", res.asString(US_ASCII));
  
  		assertEquals(0, counter.value);
  		await(cfq.put(wrapAscii("this should not be appended because the take will be completed immediately")));
  		assertEquals(1, counter.value);
  
  		cfq.close();
  	}
  }
  