/*
 * Copyright 2016-2022 chronicle.software
 *
 *       https://chronicle.software
 *
 * Licensed under Contract only
 */

package net.openhft.chronicle.wire.channel;

import net.openhft.chronicle.bytes.Bytes;
import net.openhft.chronicle.bytes.MethodReader;
import net.openhft.chronicle.core.Jvm;
import net.openhft.chronicle.wire.DocumentContext;
import net.openhft.chronicle.wire.channel.echo.DummyData;
import net.openhft.chronicle.wire.channel.echo.EchoHandler;

import java.util.concurrent.atomic.AtomicLong;

public class PerfThroughputMain {
    static final String URL = System.getProperty("url", "tcp://:1248");
    static final int RUN_TIME = Integer.getInteger("runTime", 5);
    static final boolean METHOD_RW = Jvm.getBoolean("method.rw");

    public static void main(String[] args) {
        System.out.println("-Durl=" + URL + " " +
                "-DrunTime=" + RUN_TIME + " " +
                "-Dmethod.rw=" + METHOD_RW);
        try (ChronicleContext context = ChronicleContext.newContext(URL)) {
            EchoHandler echoHandler = new EchoHandler();
            final ChronicleChannelSupplier supplier = context.newChannelSupplier(echoHandler);
            ChronicleChannel channel = supplier.get();

            doTest("unbuffered", channel);

            echoHandler.buffered(true);
            ChronicleChannel channel2 = supplier.buffered(true).get();

            doTest("buffered", channel2);
        }
    }

    private static void doTest(String desc, ChronicleChannel channel) {
        DummyData data = new DummyData();
        for (int size = 32 << 10; size >= 8; size /= 2) {
            data.data(new byte[size - Long.BYTES]);
            long start = System.currentTimeMillis();
            long end = start + RUN_TIME * 1000L;
            final Echoing echoing = channel.methodWriter(Echoing.class);
            long count = 0;
            AtomicLong unread = new AtomicLong(0);
            int window = (128 << 20) / size;
            if (METHOD_RW) {
                MethodReader reader = channel.methodReader((Echoing) d -> {
                    unread.getAndDecrement();
                });
                do {
                    echoing.echo(data);
                    unread.getAndIncrement();
                    count++;
                    do {
                        reader.readOne();
                    } while (unread.get() > window);
                } while (System.currentTimeMillis() < end);

                do {
                    reader.readOne();
                } while (unread.get() > 0);

            } else {
                InternalChronicleChannel icc = (InternalChronicleChannel) channel;
                do {
                    final Bytes<?> bytes = icc.acquireProducer().bytes();
                    bytes.writeInt(size);
                    for (int i = 0; i < size; i += 8)
                        bytes.writeLong(0L);
                    icc.releaseProducer();

                    unread.getAndIncrement();
                    count++;
                    do {
                        try (DocumentContext dc = channel.readingDocument()) {
                            if (dc.isPresent())
                                unread.getAndDecrement();
                        }
                    } while (unread.get() > window);
                } while (System.currentTimeMillis() < end);

                do {
                    try (DocumentContext dc = channel.readingDocument()) {
                        if (dc.isPresent())
                            unread.getAndDecrement();
                    }
                } while (unread.get() > 0);
            }
            long time = System.currentTimeMillis() - start;
            long totalBytes = size * count;
            long MBps = totalBytes / time / (1_000_000 / 1_000);
            long rate = count * 1000 / time;
            System.out.println("desc: " + desc + ", size: " + size + ", MBps: " + MBps + ", mps: " + rate);
        }
    }
}

@@ -1169,32 +1169,77 @@
 r);%0A            
+echoHandler.buffered(true);%0A            try (
 ChronicleChannel
@@ -1246,16 +1246,17 @@
  channel
+2
  = suppl
@@ -1259,24 +1259,44 @@
 upplier.
-get();%0A%0A
+buffered(true).get()) %7B%0A    
         
@@ -1307,18 +1307,16 @@
 doTest(%22
-un
 buffered
@@ -1325,18 +1325,33 @@
  channel
+2
 );
+%0A            %7D
 %0A%0A      
@@ -1369,35 +1369,36 @@
 andler.buffered(
-tru
+fals
 e);%0A            
@@ -1397,16 +1397,21 @@
         
+try (
 Chronicl
@@ -1418,33 +1418,32 @@
 eChannel channel
-2
  = supplier.buff
@@ -1442,31 +1442,21 @@
 ier.
-buffered(true).get();%0A%0A
+get()) %7B%0A    
     
@@ -1463,32 +1463,34 @@
         doTest(%22
+un
 buffered%22, chann
@@ -1483,36 +1483,50 @@
 ffered%22, channel
-2
 );%0A
+            %7D%0A%0A
         %7D%0A    %7D%0A

Add a single channel throughput test