package net.openhft.chronicle.queue;

import net.openhft.chronicle.bytes.Bytes;
import net.openhft.chronicle.queue.impl.single.SingleChronicleQueueBuilder;
import net.openhft.chronicle.threads.NamedThreadFactory;
import net.openhft.chronicle.wire.WireType;
import org.junit.Test;

import java.io.File;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author Rob Austin.
 */
public class ThreadedQueueTest {

    private static final int BLOCK_SIZE = 256 << 20;
    public static final int MESSAGE_SIZE = 1024;
    public static final int REQUIRED_COUNT = 100_000;

    @Test(timeout = 5000)
    public void testName() throws Exception {

        final String path = ChronicleQueueTestBase.getTmpDir() + "/deleteme.q";

        new File(path).deleteOnExit();

        final Bytes bytes = Bytes.elasticByteBuffer();
        final AtomicInteger counter = new AtomicInteger();
        final ChronicleQueue rqueue = new SingleChronicleQueueBuilder(path)
                .wireType(WireType.FIELDLESS_BINARY)
                .blockSize(BLOCK_SIZE)
                .build();

        final ExcerptTailer tailer = rqueue.createTailer();

        final ChronicleQueue wqueue = new SingleChronicleQueueBuilder(path)
                .wireType(WireType.FIELDLESS_BINARY)
                .blockSize(BLOCK_SIZE)
                .build();

        final ExcerptAppender appender = wqueue.createAppender();
        final Bytes message = Bytes.elasticByteBuffer();


        Executors.newSingleThreadExecutor(new NamedThreadFactory("tailer", true)).submit(() -> {

            while (counter.get() < REQUIRED_COUNT) {
                bytes.clear();

                if (tailer.readBytes(bytes)) {
                    System.out.println(bytes);
                    counter.incrementAndGet();
                } else {
                    Thread.yield();
                }

            }
        });

        Executors.newSingleThreadExecutor(new NamedThreadFactory("appender", true)).submit(() -> {
            for (int i = 0; i < REQUIRED_COUNT; i++) {
                message.clear();
                message.append(i);
                appender.writeBytes(message);
            }
        });


        while (counter.get() < REQUIRED_COUNT) {
            Thread.yield();
        }

    }


    @Test(timeout = 5000)
    public void testTailerReadingEmptyQueue() throws Exception {

        final String path = ChronicleQueueTestBase.getTmpDir() + "/deleteme.q";

        new File(path).deleteOnExit();


        final AtomicInteger counter = new AtomicInteger();
        final ChronicleQueue rqueue = new SingleChronicleQueueBuilder(path)
                .wireType(WireType.FIELDLESS_BINARY)
                .blockSize(BLOCK_SIZE)
                .build();

        final ExcerptTailer tailer = rqueue.createTailer();

        final ChronicleQueue wqueue = new SingleChronicleQueueBuilder(path)
                .wireType(WireType.FIELDLESS_BINARY)
                .blockSize(BLOCK_SIZE)
                .build();


        Bytes bytes = Bytes.elasticByteBuffer();
        tailer.readBytes(bytes);

        final ExcerptAppender appender = wqueue.createAppender();
        appender.writeBytes(Bytes.wrapForRead("Hello World".getBytes()));

        bytes.clear();
        tailer.readBytes(bytes);

        //   System.out.println("");


    }

}

@@ -454,61 +454,8 @@
  %7B%0A%0A
-    private static final int BLOCK_SIZE = 256 %3C%3C 20;%0A
     
@@ -552,16 +552,69 @@
 100_000;
+%0A    private static final int BLOCK_SIZE = 256 %3C%3C 20;
 %0A%0A    @T
@@ -1717,24 +1717,26 @@
 s(bytes)) %7B%0A
+//
             
@@ -1813,24 +1813,26 @@
 ntAndGet();%0A
+//
             
@@ -1844,16 +1844,18 @@
  else %7B%0A
+//
         
@@ -1907,32 +1907,83 @@
 %0A%0A            %7D%0A
+            System.out.println(%22Read %22 + counter);%0A
         %7D);%0A%0A   

Use an ordered write