class PublishHandler extends AbstractHandler<PublishHandler> {
    private String publish;
    private SyncMode syncMode;
    private int publishSourceId = 0;

    static void copyFromChannelToQueue(ChronicleChannel channel, Pauser pauser, ChronicleQueue publishQueue, SyncMode syncMode) {<FILL_FUNCTION_BODY>}

    private static void syncAppender(ExcerptAppender appender, SyncMode syncMode) {
        if (syncMode == SyncMode.SYNC) {
            try (DocumentContext dc2 = appender.writingDocument()) {
                dc2.wire().write("sync").text("");
            }
        }
        appender.sync();
    }

    public String publish() {
        return publish;
    }

    public PublishHandler publish(String publish) {
        this.publish = publish;
        return this;
    }

    public SyncMode syncMode() {
        return syncMode;
    }

    public PublishHandler syncMode(SyncMode syncMode) {
        this.syncMode = syncMode;
        return this;
    }

    public int publishSourceId() {
        return publishSourceId;
    }

    public PublishHandler publishSourceId(int publishSourceId) {
        this.publishSourceId = publishSourceId;
        return this;
    }

    @Override
    public void run(ChronicleContext context, ChronicleChannel channel) {
        Pauser pauser = Pauser.balanced();

        Thread.currentThread().setName("publish~reader");
        try (AffinityLock lock = context.affinityLock()) {
            copyFromChannelToQueue(channel, pauser, newQueue(context, publish, syncMode, publishSourceId), syncMode);
        }
    }

    @Override
    public ChronicleChannel asInternalChannel(ChronicleContext context, ChronicleChannelCfg channelCfg) {
        return new PublishQueueChannel(channelCfg, this, newQueue(context, publish, syncMode, publishSourceId));
    }
}

        try (ChronicleQueue publishQ = publishQueue;
             ExcerptAppender appender = publishQ.createAppender()) {
            appender.singleThreadedCheckDisabled(true);  // assume we are thread safe

            boolean needsSync = false;
            while (!channel.isClosed()) {
                try (DocumentContext dc = channel.readingDocument()) {
                    pauser.unpause();

                    if (!dc.isPresent()) {
                        if (needsSync) {
                            syncAppender(appender, syncMode);
                            needsSync = false;
                        }
                        continue;
                    }
                    if (dc.isMetaData()) {
                        // read message
                        continue;
                    }

                    try (DocumentContext dc2 = appender.writingDocument()) {
                        dc.wire().copyTo(dc2.wire());
                        needsSync = syncMode == SyncMode.SYNC || syncMode == SyncMode.ASYNC;
                    }
                }
            }

        } finally {
            Thread.currentThread().setName("connections");
        }
    
<methods>public void <init>() ,public java.lang.Boolean buffered() ,public net.openhft.chronicle.queue.channel.PublishHandler buffered(java.lang.Boolean) <variables>private java.lang.Boolean buffered