<commit_before>package net.openhft.chronicle.network.connection;

import net.openhft.chronicle.core.Jvm;
import net.openhft.chronicle.core.io.Closeable;
import net.openhft.chronicle.wire.WireOut;
import net.openhft.chronicle.wire.Wires;
import net.openhft.chronicle.wire.YamlLogging;
import org.jetbrains.annotations.NotNull;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Queue;
import java.util.concurrent.LinkedTransferQueue;
import java.util.function.Consumer;

/**
 * Created by peter.lawrey on 09/07/2015.
 */
public class WireOutPublisher implements Closeable {
    private static final int WARN_QUEUE_LENGTH = 50;
    private final Queue<Consumer<WireOut>> publisher = new LinkedTransferQueue<>();
    private static final Logger LOG = LoggerFactory.getLogger(WireOutPublisher.class);
    private volatile boolean closed;

    /**
     * Apply waiting messages and return false if there was none.
     *
     * @param out buffer to write to.
     */
    public void applyAction(@NotNull WireOut out, @NotNull Runnable runnable) {
        if (publisher.isEmpty()) {
            synchronized (this) {
                runnable.run();
            }
        }
        while (out.bytes().writePosition() < out.bytes().realCapacity() / 4) {
            Consumer<WireOut> wireConsumer = publisher.poll();
            if (wireConsumer == null)
                break;
            wireConsumer.accept(out);


            if (Jvm.IS_DEBUG && YamlLogging.showServerWrites)
                try {


                    LOG.info("\nServer Publishes (from async publisher ) :\n" +
                            Wires.fromSizePrefixedBinaryToText(out.bytes()));

                } catch (Exception e) {
                    LOG.info("\nServer Publishes ( from async publisher - corrupted ) :\n" +
                            out.bytes().toDebugString());
                    e.printStackTrace();
                }
        }
    }

    public void add(Consumer<WireOut> outConsumer) {

        if (closed) {
            throw new IllegalStateException("Closed");

        } else {
            int size = publisher.size();
            if (size > WARN_QUEUE_LENGTH)
                System.out.println("publish length: " + size);

            publisher.add(outConsumer);
        }
    }

    public boolean isClosed() {
        return closed;
    }

    @Override<commit_msg>changed log level to debug<commit_after>package net.openhft.chronicle.network.connection;

import net.openhft.chronicle.core.Jvm;
import net.openhft.chronicle.core.io.Closeable;
import net.openhft.chronicle.wire.WireOut;
import net.openhft.chronicle.wire.Wires;
import net.openhft.chronicle.wire.YamlLogging;
import org.jetbrains.annotations.NotNull;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Queue;
import java.util.concurrent.LinkedTransferQueue;
import java.util.function.Consumer;

/**
 * Created by peter.lawrey on 09/07/2015.
 */
public class WireOutPublisher implements Closeable {
    private static final int WARN_QUEUE_LENGTH = 50;
    private final Queue<Consumer<WireOut>> publisher = new LinkedTransferQueue<>();
    private static final Logger LOG = LoggerFactory.getLogger(WireOutPublisher.class);
    private volatile boolean closed;

    /**
     * Apply waiting messages and return false if there was none.
     *
     * @param out buffer to write to.
     */
    public void applyAction(@NotNull WireOut out, @NotNull Runnable runnable) {
        if (publisher.isEmpty()) {
            synchronized (this) {
                runnable.run();
            }
        }
        while (out.bytes().writePosition() < out.bytes().realCapacity() / 4) {
            Consumer<WireOut> wireConsumer = publisher.poll();
            if (wireConsumer == null)
                break;
            wireConsumer.accept(out);


            if (Jvm.IS_DEBUG && YamlLogging.showServerWrites)
                try {


                    LOG.info("\nServer Publishes (from async publisher ) :\n" +
                            Wires.fromSizePrefixedBinaryToText(out.bytes()));

                } catch (Exception e) {
                    LOG.info("\nServer Publishes ( from async publisher - corrupted ) :\n" +
                            out.bytes().toDebugString());
                    e.printStackTrace();
                }
        }
    }

    public void add(Consumer<WireOut> outConsumer) {

        if (closed) {
            throw new IllegalStateException("Closed");

        } else {
            int size = publisher.size();
            if (size > WARN_QUEUE_LENGTH)
                LOG.debug("publish length: " + size);

            publisher.add(outConsumer);
        }
    }

    public boolean isClosed() {
        return closed;
    }

    @Override
{'fraction_non_alphanumeric': 0.10406197654941374, 'fraction_numerical': 0.005443886097152429, 'mean_word_length': 2.1677718832891246, 'pattern_counts': {'":': 0, '<': 15, '<?xml version=': 0, '>': 15, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '421413', 'n_tokens_mistral': 1377, 'n_tokens_neox': 1270, 'n_words': 295}