package net.openhft.chronicle.queue;

import net.openhft.chronicle.core.Jvm;
import net.openhft.chronicle.core.OS;
import net.openhft.chronicle.wire.DocumentContext;
import org.junit.Assert;
import org.junit.Assume;
import org.junit.Test;

import java.io.BufferedReader;
import java.io.File;
import java.io.IOException;
import java.io.InputStreamReader;

import static org.junit.Assert.fail;
import static org.junit.Assume.assumeFalse;

public class CheckHalfWrittenMsgNotSeenByTailerTest extends QueueTestCommon {
    static {
        // load the lass
        HalfWriteAMessage.class.getName();
    }

    public static class HalfWriteAMessage {

        // writes three messages the third messas is half written
        public static void main(String[] args) throws InterruptedException {
            writeIncompleteMessage(args[0], true);
        }

        private static void writeIncompleteMessage(String arg, boolean exit) throws InterruptedException {
            System.out.println("half writing a message to " + arg);

            try( final ChronicleQueue single = ChronicleQueue.single(arg) ) {
                final ExcerptAppender excerptAppender = single.acquireAppender();

                try (final DocumentContext dc = excerptAppender.writingDocument()) {
                    dc.wire().write("key1").text("hello world 1");
                    dc.wire().write("key2").text("hello world 2");
                }

                try (final DocumentContext dc = excerptAppender.writingDocument()) {
                    dc.wire().write("key1").text("hello world 3");
                    dc.wire().write("key2").text("hello world 4");
                }

                DocumentContext dc = excerptAppender.writingDocument();
                dc.wire().write("key1").text("hello world 5");

                // give time to flush
                Thread.sleep(1);

                System.out.println("== FINISHED WRITING DATA ==");

                // this will create a half written message, as we are going to system exit
                if (exit)
                    System.exit(-1);

                dc.wire().write("key2").text("hello world 6");
            }
        }
    }

    @Test
    public void checkTailerOnlyReadsTwoMessageOneProcess() throws InterruptedException {
        Assume.assumeTrue(!OS.isWindows());
        final File queueDirectory = DirectoryUtils.tempDir("halfWritten");

        HalfWriteAMessage.writeIncompleteMessage(queueDirectory.toString(), false);
        for (int i = 0; i < 3; i++) {
            System.gc();
            Jvm.pause(50);
        }

        try (final ChronicleQueue single = ChronicleQueue.single(queueDirectory.getPath());
             final ExcerptTailer tailer = single.createTailer()) {

            try (final DocumentContext dc = tailer.readingDocument()) {
                Assert.assertTrue(dc.isPresent());
                Assert.assertEquals("hello world 1", dc.wire().read("key1").text());
                Assert.assertEquals("hello world 2", dc.wire().read("key2").text());
            }

            try (final DocumentContext dc = tailer.readingDocument()) {
                Assert.assertTrue(dc.isPresent());
                Assert.assertEquals("hello world 3", dc.wire().read("key1").text());
                Assert.assertEquals("hello world 4", dc.wire().read("key2").text());
            }

            try (final DocumentContext dc = tailer.readingDocument()) {
                final boolean present = dc.isPresent();
                if (present) {
                    System.out.println(dc.wire().bytes().toHexString());
                    String key = dc.wire().readEvent(String.class);
                    String value = dc.wire().getValueIn().text();
                    fail("key: " + key + ", value: " + value);
                }
            }
        }
    }

    @Test
    public void checkTailerOnlyReadsTwoMessageTwoProcesses() throws IOException, InterruptedException {
        Assume.assumeTrue(!OS.isWindows());
        Assume.assumeTrue(!OS.isMacOSX());
        final File queueDirectory = DirectoryUtils.tempDir("halfWritten");

        final String command = String.format("mvn compile exec:java -Dexec.classpathScope=test " +
                "-Dexec.mainClass=%s -Dexec.args=\"%s\"", HalfWriteAMessage.class.getName(), queueDirectory.getAbsoluteFile());
        runCommand(command);

        try (final ChronicleQueue single = ChronicleQueue.single(queueDirectory.getPath());
             final ExcerptTailer tailer = single.createTailer()) {

            try (final DocumentContext dc = tailer.readingDocument()) {
                Assert.assertTrue(dc.isPresent());
                Assert.assertEquals("hello world 1", dc.wire().read("key1").text());
                Assert.assertEquals("hello world 2", dc.wire().read("key2").text());
            }

            try (final DocumentContext dc = tailer.readingDocument()) {
                Assert.assertTrue(dc.isPresent());
                Assert.assertEquals("hello world 3", dc.wire().read("key1").text());
                Assert.assertEquals("hello world 4", dc.wire().read("key2").text());
            }

            try (final DocumentContext dc = tailer.readingDocument()) {
                final boolean present = dc.isPresent();
                if (present) {
                    Jvm.error().on(getClass(), "Found an excerpt " + dc.wire().bytes().toHexString());

                    String key = dc.wire().readEvent(String.class);
                    String value = dc.wire().getValueIn().text();
                    fail("key: " + key + ", value: " + value);
                }
            }
        }
    }

    private static void runCommand(String command) throws IOException, InterruptedException {
        Process p = Runtime.getRuntime().exec(command);
        BufferedReader stdInput = new BufferedReader(new
                InputStreamReader(p.getInputStream()));

        BufferedReader stdError = new BufferedReader(new
                InputStreamReader(p.getErrorStream()));

        boolean buildFailure = false;
        String s;
        // read the output from the command
        //      System.out.println("Here is the standard output of the command:\n");
        while ((s = stdInput.readLine()) != null) {

            System.out.println(s);
            if (s.contains("BUILD FAILURE"))
                buildFailure = true;

            // wait for Replication Started
            if ("== FINISHED WRITING DATA ==".equals(s))
                return;

        }

        // read any errors from the attempted command
        System.out.println("Here is the standard error of the command (if any):\n");
        while ((s = stdError.readLine()) != null) {
            System.out.println(s);
        }
        p.waitFor();
        // test can't be run this way.
        assumeFalse(buildFailure);
    }
}

@@ -108,16 +108,82 @@
 ore.OS;%0A
+import net.openhft.chronicle.testframework.process.ProcessRunner;%0A
 import n
@@ -4197,157 +4197,42 @@
     
-final String command = String.format(%22mvn compile exec:java -Dexec.classpathScope=test %22 +%0A                %22-Dexec.mainClass=%25s -Dexec.args=%5C%22%25s%5C%22%22, 
+runCommand(ProcessRunner.runClass(
 Half
@@ -4250,26 +4250,16 @@
 ge.class
-.getName()
 , queueD
@@ -4282,43 +4282,15 @@
 lute
-File());%0A        runCommand(command
+Path())
 );%0A%0A
@@ -5613,22 +5613,17 @@
 and(
-String command
+Process p
 ) th
@@ -5667,64 +5667,8 @@
 n %7B%0A
-        Process p = Runtime.getRuntime().exec(command);%0A
     
@@ -5895,46 +5895,8 @@
 );%0A%0A
-        boolean buildFailure = false;%0A
     
@@ -6129,90 +6129,8 @@
 (s);
-%0A            if (s.contains(%22BUILD FAILURE%22))%0A                buildFailure = true;
 %0A%0A  
@@ -6492,32 +6492,32 @@
 n(s);%0A        %7D%0A
+
         p.waitFo
@@ -6525,82 +6525,8 @@
 ();%0A
-        // test can't be run this way.%0A        assumeFalse(buildFailure);%0A
     

Fix flaky test which poisoned Maven repository and target