FixSocketReader { public boolean readMessages() throws IOException { readMoreData(buffer); if (buffer.position() < MIN_FIX_MSG_SIZE) return false; int end = buffer.position(), pos = end; if (buffer.get(--pos) != END_OF_FIX_MESSAGE) { do { if (pos <= 0) return false; } while (buffer.get(--pos) != END_OF_FIX_MESSAGE); } do { int pos2 = pos; while (buffer.get(--pos2) != END_OF_FIX_MESSAGE) if (pos2 <= MIN_FIX_MSG_SIZE) return false; if (buffer.get(pos2 + 1) == '1' && buffer.get(pos2 + 2) == '0' && buffer.get(pos2 + 3) == '=') { pos++; excerpt.startExcerpt(1 + 8 + 4 + pos); excerpt.writeStopBit(pos); buffer.position(0); buffer.limit(pos); excerpt.write(buffer); excerpt.writeLong(currentTimeNS()); excerpt.finish(); buffer.limit(end); buffer.position(pos); if (buffer.remaining() > 0) buffer.compact(); else buffer.clear(); return true; } pos = pos2; } while (true); } FixSocketReader(SocketChannel sc, Excerpt excerpt); boolean readMessages(); void close(); boolean isClosed(); Excerpt excerpt();  }
@Test public void testReadMessages() throws Exception { String path = TMP + "/testReadMessages"; ChronicleTools.deleteOnExit(path); IndexedChronicle chronicle = new IndexedChronicle(path); Excerpt excerpt = chronicle.createExcerpt(); FixSocketReader fsr = new FixSocketReader(null, excerpt) { final byte[] bytes = S.getBytes("US-ASCII"); int start = 0; @Override protected void readMoreData(@NotNull ByteBuffer buffer) throws IOException { int end = Math.min(bytes.length, start + 256); buffer.put(bytes, start, end - start); start = end; if (end == bytes.length) close(); } }; FixDecoder fd = new FixDecoder(fsr); FixDecodeListener listener = new FixDecodeListener() { @Override public void onField(int fid, @NotNull Excerpt value) { String text = value.parseUTF(StopCharTesters.FIX_TEXT); System.out.println("\t" + fid + "=" + text); } @Override public void onEndOfMessage() { System.out.println("EOM"); } @Override public void onEndOfBatch() { System.out.println("EOB"); } }; while (!fsr.isClosed()) { fd.readMessages(listener); } excerpt.index(-1); int count = 0; while (excerpt.nextIndex()) { int length = (int) excerpt.readStopBit(); String fix = ChronicleTools.asString(excerpt, excerpt.position(), length); assertTrue(fix.startsWith("8=FIX")); assertTrue(fix.endsWith(".")); assertEquals(8, fix.length() - fix.lastIndexOf(".10=")); int pos = -1; while ((pos = fix.indexOf(".10=", pos + 1)) > 0) count++; } assertEquals(86, count); chronicle.close(); }