import net.openhft.chronicle.ExcerptTailer; //导入依赖的package包/类
@Test
public void testTailerToEnd1() throws IOException {
  final String baseDir = getTestPath();
  assertNotNull(baseDir);

  final VanillaChronicle chronicle = new VanillaChronicle(baseDir);
  chronicle.clear();

  try {
    ExcerptAppender appender = chronicle.createAppender();
    for (long i = 0; i < 3; i++) {
      appender.startExcerpt();
      appender.writeLong(i);
      appender.finish();
    }

    // test a vanilla tailer, wind to end
    ExcerptTailer tailer = chronicle.createTailer().toEnd();
    assertEquals(2, tailer.readLong());
    assertFalse(tailer.nextIndex());

    appender.close();
    tailer.close();

    chronicle.checkCounts(1, 1);
  } finally {
    chronicle.close();
    chronicle.clear();

    assertFalse(new File(baseDir).exists());
  }
}
 

import net.openhft.chronicle.ExcerptTailer; //导入依赖的package包/类
@Test
public void testTailerToEnd2() throws IOException {
  final String baseDir = getTestPath();
  assertNotNull(baseDir);

  final VanillaChronicle wchronicle = new VanillaChronicle(baseDir);
  wchronicle.clear();

  try {
    ExcerptAppender appender = wchronicle.createAppender();
    for (long i = 0; i < 3; i++) {
      appender.startExcerpt();
      appender.writeLong(i);
      appender.finish();
    }

    // test a vanilla tailer, wind to end
    final VanillaChronicle rchronicle = new VanillaChronicle(baseDir);
    final ExcerptTailer tailer = rchronicle.createTailer().toEnd();
    assertEquals(2, tailer.readLong());
    assertFalse(tailer.nextIndex());

    appender.close();
    tailer.close();

    rchronicle.checkCounts(1, 1);
    rchronicle.close();

    wchronicle.checkCounts(1, 1);

  } finally {
    wchronicle.close();
    wchronicle.clear();

    assertFalse(new File(baseDir).exists());
  }
}
 

import net.openhft.chronicle.ExcerptTailer; //导入依赖的package包/类
private static Set<String> readAvailableValues(final ExcerptTailer tailer) {
  final Set<String> values = new TreeSet<String>();
  while (tailer.nextIndex()) {
    values.add(tailer.readUTF());
  }
  return values;
}
 

import net.openhft.chronicle.ExcerptTailer; //导入依赖的package包/类
@Test
public void testTailerToEnd1() throws IOException {
  final String baseDir = "build/vanilla";
  assertNotNull(baseDir);

  final VanillaChronicle chronicle = new VanillaChronicle(baseDir);
  chronicle.clear();

  try {
    ExcerptAppender appender = chronicle.createAppender();
    for (long i = 0; i < 3; i++) {
      appender.startExcerpt();
      appender.writeLong(i);
      appender.finish();
    }
    appender.close();
    // test a vanilla tailer, wind to end
    ExcerptTailer tailer = chronicle.createTailer().toEnd();
    assertEquals(2, tailer.readLong());
    assertFalse(tailer.nextIndex());

    tailer.close();

    chronicle.checkCounts(1, 1);
  } finally {
    chronicle.close();
    chronicle.clear();
  }
}
 

import net.openhft.chronicle.ExcerptTailer; //导入依赖的package包/类
@Test
public void testAppender() throws Exception {
  String dataDir = "build/chronicle";
  FileUtil.removeIfExist(dataDir, true);
  System.out.println("data dir: " + dataDir);
  IndexedChronicle chronicle = new IndexedChronicle(dataDir + "/data", ChronicleConfig.SMALL);
  ExcerptAppender appender = chronicle.createAppender();
  appender.startExcerpt(8);
  appender.writeLong(10l);
  appender.finish();
  
  for(int i = 0; i < 10; i++) {
    String string = "test " + i ;
    byte[] data = string.getBytes() ;
    appender.startExcerpt(data.length);
    appender.write(data);
    appender.finish();
  }
  
  ExcerptTailer reader = chronicle.createTailer();
  reader.index(1) ;
  while (reader.nextIndex()) {
    System.out.println("Read string by reader: " + reader.readLine());
  }
  
  Excerpt excerpt = chronicle.createExcerpt() ;
  excerpt.index(1);
  while(excerpt.nextIndex()) {
    System.out.println("Read string by exerpt: " + excerpt.readLine());
  }
  
  appender.close();
  reader.close();
  excerpt.close();
  
  excerpt = chronicle.createExcerpt() ;
  long longNumber = excerpt.readLong() ;
  System.out.println("Long number = " + longNumber);
}
 

import net.openhft.chronicle.ExcerptTailer; //导入依赖的package包/类
public ExcerptTailer getTailer() {
    return tailer;
}
 

import net.openhft.chronicle.ExcerptTailer; //导入依赖的package包/类
public ChronicleReceiver(final DirectFactory directFactory, final MutableDirectBuffer mutableDirectBuffer, final ExcerptTailer tailer) {
    this.directFactory = Objects.requireNonNull(directFactory);
    this.mutableDirectBuffer = Objects.requireNonNull(mutableDirectBuffer);
    this.tailer = Objects.requireNonNull(tailer);
}
 

import net.openhft.chronicle.ExcerptTailer; //导入依赖的package包/类
@Test
public void testTailer() throws IOException {
  final int RUNS = 1000000; // 5000000;

  final String baseDir = getTestPath();
  assertNotNull(baseDir);

  final VanillaChronicleConfig config = new VanillaChronicleConfig();
  config.defaultMessageSize(128);
  config.indexBlockSize(256 << 10);
  config.dataBlockSize(512 << 10);

  final VanillaChronicle chronicle1 = new VanillaChronicle(baseDir, config);
  chronicle1.clear();

  final VanillaChronicle chronicle2 = new VanillaChronicle(baseDir, config);

  try {
    ExcerptAppender appender = chronicle1.createAppender();
    ExcerptTailer tailer = chronicle2.createTailer();

    assertEquals(-1L, tailer.index());
    for (int i = 0; i < RUNS; i++) {
      // if ((i & 65535) == 0)
      // System.err.println("i: " + i);
      // if (i == 88000)
      // Thread.yield();
      assertFalse(tailer.nextIndex());
      appender.startExcerpt();
      int value = 1000000000 + i;
      appender.append(value).append(' ');
      appender.finish();
      chronicle1.checkCounts(1, 2);
      assertTrue("i: " + i, tailer.nextIndex());
      chronicle2.checkCounts(1, 2);
      assertTrue("i: " + i + " remaining: " + tailer.remaining(), tailer.remaining() > 0);
      assertEquals("i: " + i, value, tailer.parseLong());
      assertEquals("i: " + i, 0, tailer.remaining());
      tailer.finish();
      chronicle2.checkCounts(1, 2);
    }

    appender.close();
    tailer.close();

    chronicle2.checkCounts(1, 1);
    chronicle1.checkCounts(1, 1);
  } finally {
    chronicle2.close();

    chronicle1.close();
    chronicle1.clear();

    assertFalse(new File(baseDir).exists());
  }
}
 

import net.openhft.chronicle.ExcerptTailer; //导入依赖的package包/类
@Test
public void testTailerPerf() throws IOException {
  final int WARMUP = 50000;
  final int RUNS = 5000000;

  final String baseDir = getTestPath();
  assertNotNull(baseDir);

  final VanillaChronicle chronicle = new VanillaChronicle(baseDir);
  chronicle.clear();

  try {
    final ExcerptAppender appender = chronicle.createAppender();
    final ExcerptTailer tailer = chronicle.createTailer();
    long start = 0;
    assertEquals(-1L, tailer.index());
    for (int i = -WARMUP; i < RUNS; i++) {
      if (i == 0)
        start = System.nanoTime();
      boolean condition0 = tailer.nextIndex();
      if (condition0)
        assertFalse("i: " + i, condition0);
      appender.startExcerpt();
      int value = 1000000000 + i;
      appender.append(value).append(' ');
      appender.finish();
      boolean condition = tailer.nextIndex();
      long actual = tailer.parseLong();
      if (i < 0) {
        assertTrue("i: " + i, condition);
        assertEquals("i: " + i, value, actual);
      }
      tailer.finish();
    }
    long time = System.nanoTime() - start;
    System.out.printf("Average write/read times was %.3f us%n", time / RUNS / 1e3);

    appender.close();
    tailer.close();

    chronicle.checkCounts(1, 1);

  } finally {
    chronicle.close();
    chronicle.clear();

    assertFalse(new File(baseDir).exists());
  }
}
 

import net.openhft.chronicle.ExcerptTailer; //导入依赖的package包/类
@Test
public void testTailerToStart() throws IOException {
  final String baseDir = getTestPath();
  assertNotNull(baseDir);

  VanillaChronicle chronicle = new VanillaChronicle(baseDir);
  chronicle.clear();

  try {
    ExcerptAppender appender = chronicle.createAppender();
    ExcerptTailer tailer = null;

    for (long i = 0; i < 3; i++) {
      appender.startExcerpt();
      appender.writeLong(i);
      appender.finish();
    }

    // test a vanilla tailer, no rewind
    tailer = chronicle.createTailer();
    for (long i = 0; i < 3; i++) {
      assertTrue(tailer.nextIndex());
      assertEquals(i, tailer.readLong());
      tailer.finish();
    }

    // test a vanilla tailer, rewind
    tailer = chronicle.createTailer().toStart();
    for (long i = 0; i < 3; i++) {
      assertTrue(tailer.nextIndex());
      assertEquals(i, tailer.readLong());
      tailer.finish();
    }

    appender.close();
    tailer.close();

    chronicle.checkCounts(1, 1);
  } finally {
    chronicle.close();
    chronicle.clear();

    assertFalse(new File(baseDir).exists());
  }
}
 

import net.openhft.chronicle.ExcerptTailer; //导入依赖的package包/类
@Test
public void testTailerEndStart1() throws IOException {
  final String baseDir = getTestPath();
  assertNotNull(baseDir);

  final VanillaChronicle chronicle = new VanillaChronicle(baseDir);
  chronicle.clear();

  try {
    ExcerptAppender appender = chronicle.createAppender();

    // test a vanilla tailer, wind to end on an empty chronicle
    ExcerptTailer tailer = chronicle.createTailer().toEnd();
    assertFalse(tailer.nextIndex());

    // add some data to the chronicle
    for (long i = 0; i < 3; i++) {
      appender.startExcerpt();
      appender.writeLong(i);
      appender.finish();
    }

    // test that the tailer now can tail
    for (long i = 0; i < 3; i++) {
      assertTrue(tailer.nextIndex());
      assertEquals(i, tailer.readLong());
      tailer.finish();
    }

    // test a vanilla tailer, wind to end
    tailer = chronicle.createTailer().toEnd();
    assertEquals(2, tailer.readLong());
    assertFalse(tailer.nextIndex());
    tailer.finish();

    // test a vanilla tailer, rewind
    tailer = chronicle.createTailer().toStart();
    for (long i = 0; i < 3; i++) {
      assertTrue(tailer.nextIndex());
      assertEquals(i, tailer.readLong());
      tailer.finish();
    }

    appender.close();
    tailer.close();

    chronicle.checkCounts(1, 1);
  } finally {
    chronicle.close();
    chronicle.clear();

    assertFalse(new File(baseDir).exists());
  }
}
 

import net.openhft.chronicle.ExcerptTailer; //导入依赖的package包/类
@Test
public void testTailerEndStart2() throws IOException {
  final String baseDir = getTestPath();
  assertNotNull(baseDir);

  final VanillaChronicle wchronicle = new VanillaChronicle(baseDir);
  final VanillaChronicle rchronicle = new VanillaChronicle(baseDir);

  wchronicle.clear();

  try {
    ExcerptAppender appender = wchronicle.createAppender();

    // test a vanilla tailer, wind to end on an empty chronicle
    ExcerptTailer tailer = rchronicle.createTailer().toEnd();
    assertFalse(tailer.nextIndex());

    // add some data to the chronicle
    for (long i = 0; i < 3; i++) {
      appender.startExcerpt();
      appender.writeLong(i);
      appender.finish();
    }

    // test that the tailer now can tail
    for (long i = 0; i < 3; i++) {
      assertTrue(tailer.nextIndex());
      assertEquals(i, tailer.readLong());
      tailer.finish();
    }

    // test a vanilla tailer, wind to end
    tailer = rchronicle.createTailer().toEnd();
    assertEquals(2, tailer.readLong());
    assertFalse(tailer.nextIndex());
    tailer.finish();

    // test a vanilla tailer, rewind
    tailer = rchronicle.createTailer().toStart();
    for (long i = 0; i < 3; i++) {
      assertTrue(tailer.nextIndex());
      assertEquals(i, tailer.readLong());
      tailer.finish();
    }

    appender.close();
    tailer.close();

    rchronicle.checkCounts(1, 1);
    wchronicle.checkCounts(1, 1);
  } finally {
    rchronicle.close();

    wchronicle.close();
    wchronicle.clear();

    assertFalse(new File(baseDir).exists());
  }
}
 

import net.openhft.chronicle.ExcerptTailer; //导入依赖的package包/类
@Test
public void testReplicationWithRollingFilesEverySecond() throws Exception {
  // TODO int RUNS = 100000;
  final int RUNS = 5 * 1000;

  final String baseDir = getTestPath();
  assertNotNull(baseDir);

  final VanillaChronicleConfig config = new VanillaChronicleConfig();
  config.entriesPerCycle(1L << 16);
  config.cycleLength(1000, false);
  config.cycleFormat("yyyyMMddHHmmss");
  config.indexBlockSize(16L << 10);

  final VanillaChronicle chronicle = new VanillaChronicle(baseDir, config);
  chronicle.clear();

  try {
    ExcerptAppender appender = chronicle.createAppender();
    ExcerptTailer tailer = chronicle.createTailer();
    long lastMajor = 0;
    for (int i = 0; i < RUNS; i++) {
      appender.startExcerpt();
      long value = 1000000000 + i;
      appender.append(value).append(' ');
      appender.finish();
      // System.out.println("Sleeping " +i );
      Thread.sleep(1);

      assertTrue(tailer.nextIndex());
      long major = tailer.index() / config.entriesPerCycle();
      if (lastMajor == 0 || lastMajor == major) {
        // ok.
      } else if (lastMajor + 1 == major) {
        System.out.println("Major: " + major);
      } else {
        assertEquals("major jumped", lastMajor + 1, major);
      }
      lastMajor = major;
      // System.out.printf("Index: %x%n", major);
      assertEquals("i: " + i, value, tailer.parseLong());
      assertEquals("i: " + i, 0, tailer.remaining());
      tailer.finish();
    }

    appender.close();
    tailer.close();

    chronicle.checkCounts(1, 1);
  } finally {
    chronicle.close();
    chronicle.clear();

    assertFalse(new File(baseDir).exists());
  }
}
 

import net.openhft.chronicle.ExcerptTailer; //导入依赖的package包/类
@Test
public void testMultipleCycles() throws Exception {
  final String baseDir = getTestPath();
  System.out.println("base dir: " + baseDir);
  assertNotNull(baseDir);

  // Create with small data and index sizes so that the test frequently
  // generates new files
  final VanillaChronicleConfig config = new VanillaChronicleConfig();
  config.entriesPerCycle(1L << 20); // avoid overflow of the entry indexes
  config.cycleLength(1000, false); // 1 second
  config.cycleFormat("yyyyMMddHHmmss");
  config.dataBlockSize(128);
  config.indexBlockSize(64);

  final VanillaChronicle chronicle = new VanillaChronicle(baseDir, config);
  chronicle.clear();

  try {
    final ExcerptAppender appender = chronicle.createAppender();
    appendValues(appender, 1, 20);

    // Ensure the appender writes in another cycle from the initial writes
    Thread.sleep(2000L);
    appendValues(appender, 20, 40);

    // Verify that all values are read by the tailer
    final ExcerptTailer tailer = chronicle.createTailer();
    assertEquals(createRangeDataSet(1, 40), readAvailableValues(tailer));

    // Verify that the tailer reads no new data from a new cycle
    Thread.sleep(2000L);
    assertTrue(!tailer.nextIndex());

    // ### Throws java.lang.NullPointerException
    // - lastIndexFile is set to null by the previous call to nextIndex
    assertTrue(!tailer.nextIndex());

    // Append data in this new cycle
    appendValues(appender, 41, 60);

    // Verify that the tailer can read the new data
    assertEquals(createRangeDataSet(41, 60), readAvailableValues(tailer));

    appender.close();
    tailer.close();

    chronicle.checkCounts(1, 1);
  } finally {
    chronicle.close();
    chronicle.clear();

    assertFalse(new File(baseDir).exists());
  }
}
 

import net.openhft.chronicle.ExcerptTailer; //导入依赖的package包/类
@Test
public void testMultipleCycles2() throws Exception {
  final String baseDir = getTestPath();
  assertNotNull(baseDir);

  // Create with small data and index sizes so that the test frequently
  // generates new files
  final VanillaChronicleConfig config = new VanillaChronicleConfig();
  config.entriesPerCycle(1L << 20); // avoid overflow of the entry indexes
  config.cycleLength(1000, false); // 1 second
  config.cycleFormat("yyyyMMddHHmmss");
  config.dataBlockSize(128);
  config.indexBlockSize(64);

  final VanillaChronicle chronicle = new VanillaChronicle(baseDir, config);
  chronicle.clear();

  try {
    final ExcerptAppender appender = chronicle.createAppender();
    final ExcerptTailer tailer = chronicle.createTailer();
    
    // Append a small number of events in this cycle
    appendValues(appender, 1, 5);

    // Ensure the appender writes in another cycle from the initial writes
    Thread.sleep(2000L);

    appendValues(appender, 5, 50);

    // ### Fails because it only reads the values written in the first cycle
    assertEquals(createRangeDataSet(1, 50), readAvailableValues(tailer));

    appender.close();
    tailer.close();

    chronicle.checkCounts(1, 1);
  } finally {
    chronicle.close();
    chronicle.clear();

    assertFalse(new File(baseDir).exists());
  }
}
 

import net.openhft.chronicle.ExcerptTailer; //导入依赖的package包/类
@Test
public void testMultipleCycles2() throws Exception {
  final String baseDir = "build/vanilla";
  assertNotNull(baseDir);
  FileUtil.removeIfExist(baseDir, false);

  // Create with small data and index sizes so that the test frequently
  // generates new files
  VanillaChronicleConfig config = new VanillaChronicleConfig();
  config.cycleFormat("yyyy-MM-dd-HH:mm:ss");
  config.cycleLength(60 * 1000, false);
  config.entriesPerCycle(512 * 1024);
  config.dataBlockSize(4 * 1024 * 1024);
  config.indexBlockSize(1 * 1024 * 1024);
  final VanillaChronicle chronicle = new VanillaChronicle(baseDir, config);
  chronicle.clear();
  
  try {
    ExcerptAppender appender = chronicle.createAppender();
    long start = System.currentTimeMillis() ;
    for(int j = 0; j < 4500000; j++) {
      appender.startExcerpt();
      appender.writeInt(j + 1);
      appender.finish();
    }
    long exec = System.currentTimeMillis() - start ;
    System.out.println("Append in " + exec + "ms");
    
    
    for(int i = 0; i < 5; i++) {
      long sum = 0 ;
      AtomicInteger count = new AtomicInteger() ;
      ExcerptTailer tailer     = chronicle.createTailer();
      int lastValue = 0 ;
      while(tailer.nextIndex()) {
        count.incrementAndGet() ;
        lastValue = tailer.readInt();
        sum += lastValue;
        tailer.finish();
      }
      //tailer.flush() ;
      tailer.close();
      System.out.println("files = " +tailer.file()) ;
      System.out.println(
          "count = " + count.get() + ", sum = " + sum + ", last value = " + lastValue + ", last index = " + chronicle.lastIndex() + 
          ", was padding = " + tailer.wasPadding() + ", nextIndex = " + tailer.nextIndex() + 
          ", finished = " + tailer.isFinished());
    }
    
    appender.close();
    chronicle.checkCounts(1, 1);
  } finally {
    chronicle.close();
  }
}
 

import net.openhft.chronicle.ExcerptTailer; //导入依赖的package包/类
@Test
public void testReplicationWithRollingFilesEverySecond() throws Exception {
  // TODO int RUNS = 100000;
  final int RUNS = 5 * 1000;

  final String baseDir = "build/vanilla";
  assertNotNull(baseDir);

  final VanillaChronicleConfig config = new VanillaChronicleConfig();
  config.entriesPerCycle(1L << 16);
  config.cycleLength(1000, false);
  config.cycleFormat("yyyyMMddHHmmss");
  config.indexBlockSize(16L << 10);

  final VanillaChronicle chronicle = new VanillaChronicle(baseDir, config);
  chronicle.clear();
  try {
    ExcerptAppender appender = chronicle.createAppender();
    ExcerptTailer tailer = chronicle.createTailer();
    long lastMajor = 0;
    for (int i = 0; i < RUNS; i++) {
      appender.startExcerpt();
      long value = 1000000000 + i;
      appender.append(value).append(' ');
      appender.finish();
      // System.out.println("Sleeping " +i );
      Thread.sleep(1);
      assertTrue(tailer.nextIndex());
      long major = tailer.index() / config.entriesPerCycle();
      if (lastMajor == 0 || lastMajor == major) {
        // ok.
      } else if (lastMajor + 1 == major) {
        System.out.println("Major: " + major);
      } else {
        assertEquals("major jumped", lastMajor + 1, major);
      }
      lastMajor = major;
      // System.out.printf("Index: %x%n", major);
      assertEquals("i: " + i, value, tailer.parseLong());
      assertEquals("i: " + i, 0, tailer.remaining());
      tailer.finish();
    }

    appender.close();
    tailer.close();

    chronicle.checkCounts(1, 1);
  } finally {
    chronicle.close();
    //chronicle.clear();
    //assertFalse(new File(baseDir).exists());
  }
}
 

import net.openhft.chronicle.ExcerptTailer; //导入依赖的package包/类
/**
 * Initializes the consumer using the provided input
 * @param queueId
 * @param queueReader
 * @param waitStrategy
 */
public DefaultStreamingMessageQueueConsumer(final String queueId, final ExcerptTailer queueReader, final StreamingMessageQueueWaitStrategy waitStrategy) {
	this.queueId = queueId;
	this.queueReader = queueReader;
	this.waitStrategy = waitStrategy;
}