<commit_before>/*
 *  Licensed under the Apache License, Version 2.0 (the "License");
 *  you may not use this file except in compliance with the License.
 *  You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 *  Unless required by applicable law or agreed to in writing, software
 *  distributed under the License is distributed on an "AS IS" BASIS,
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  See the License for the specific language governing permissions and
 *  limitations under the License.
 *
 */
package dk.statsbiblioteket.summa.ingest.split;

import dk.statsbiblioteket.summa.common.Record;
import dk.statsbiblioteket.summa.common.configuration.Configuration;
import dk.statsbiblioteket.summa.common.filter.Payload;
import dk.statsbiblioteket.summa.common.filter.PayloadQueue;
import dk.statsbiblioteket.util.qa.QAInfo;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import java.util.NoSuchElementException;
import java.util.concurrent.TimeUnit;

/**
 * Helper implementation of StreamParser that handles the bookkeeping of
 * threaded parsing. With threaded parsing, a thread is responsible for reading
 * from the Stream, producing Records and adding the Records to a queue.
 * Reading from the queue is done from outside the Thread.
 * </p><p>
 * Implementators of this abstract class only needs to override the
 * {@link #protectedRun(Payload)} method. They should check {@link #running}
 * frequently and stop processing if it is false.
 */
@QAInfo(level = QAInfo.Level.NORMAL,
        state = QAInfo.State.IN_DEVELOPMENT,
        author = "te")
public abstract class ThreadedStreamParser implements StreamParser {
    private static Log log = LogFactory.getLog(ThreadedStreamParser.class);

    /**
     * The maximum queue size, counted in number of Records.
     * </p><p>
     * Optional. Default is 100.
     */
    public static final String CONF_QUEUE_SIZE = "summa.ingest.stream.threadedstreamparser.queue.size";
    public static final int DEFAULT_QUEUE_SIZE = 100;

    /**
     * The maximum queue size, counted in bytes.
     * </p><p>
     * Optional. Default is 5 MB.
     */
    public static final String CONF_QUEUE_BYTESIZE = "summa.ingest.stream.threadedstreamparser.queue.bytesize";
    public static final int DEFAULT_QUEUE_BYTESIZE = 5*1000*1000; // 5 MB

    /**
     * The maximum number of milliseconds to wait for data when hasNext(),
     * next() or pump() is called.
     * </p><p>
     * Optional. Default is Integer.MAX_VALUE.
     */
    public static final String CONF_QUEUE_TIMEOUT = "summa.ingest.stream.threadedstreamparser.queue.timeout";
    public static final int DEFAULT_QUEUE_TIMEOUT = Integer.MAX_VALUE;
    private int queueTimeout = DEFAULT_QUEUE_TIMEOUT;

    //private static final long HASNEXT_SLEEP = 50; // Sleep-ms between polls
    private static final Payload INTERRUPTOR = new Payload(new Record("dummyID", "dummyStreamBase", new byte[0]));

    /*
     * Holds the produced Payloads. Generated Payloads will always be followed
     * by {@link #INTERRUPTOR}.
     */
    private PayloadQueue queue;
    /*
     * The next Payload to deliver. This is assigned by {@link #hasNext} and
     * used by {@link #next}. toDeliver will never be {@link INTERRUPTOR}.
     */
    private Payload toDeliver = null;

    /**
     * The source Payload that is to be processed.
     */
    private Payload sourcePayload;

    /**
     * If true, processing should proceed normally. If false, processing should
     * be terminated when it can be done without affecting stability.
     */
    protected boolean running = false;

    /**
     * empty == true iff no Payload has been assigned yet or
     * (the processing thread is not running and the
     * interruptor-Payload has been encountered in the queue).
     * </p><p>
     * {@link #open} must only be called when empty == true.
     */
    private boolean empty = true; // Starting condition

    private Throwable lastError = null;
    private Thread runningThread = null;
    private long queueCount = 0;

    public ThreadedStreamParser(Configuration conf) {
        queue = new PayloadQueue(conf.getInt(CONF_QUEUE_SIZE, DEFAULT_QUEUE_SIZE),
                                 conf.getInt(CONF_QUEUE_BYTESIZE, DEFAULT_QUEUE_BYTESIZE));
        queueTimeout = conf.getInt(CONF_QUEUE_TIMEOUT, queueTimeout);
        log.debug("Constructed ThreadedStreamParser with queue-size " + queue.remainingCapacity()
                  + " and queue timeout " + queueTimeout + " ms");
    }

    @Override
    public void open(Payload streamPayload) {
        //noinspection DuplicateStringLiteralInspection
        log.debug("open(" + streamPayload + ") called");
        if (!empty) {
            throw new IllegalStateException(String.format(
                    "Already parsing %s when open(%s) was called",
                    sourcePayload, streamPayload));
        }
        queue.clear(); // Clean-up from previous runs
        if (streamPayload.getStream() == null) {
            log.warn("No stream in received " + streamPayload + ". No Records will be generated");
            empty = true;
            return;
        }
        sourcePayload = streamPayload;
        startThread();
    }

    private void startThread() {
        log.trace("Starting Thread for " + sourcePayload);
        running = true;
        empty = false;
        /* Set up a thread reporting errors back to us */
        //noinspection DuplicateStringLiteralInspection
        runningThread = new Thread(new Runnable() {
            @SuppressWarnings("ObjectToString")
            @Override
            public void run() {
                log.trace("run() entered");
                try {
                    protectedRun(sourcePayload);
                    if (autoClose()) {
                        sourcePayload.close();
                    }
                } catch (Exception e) {
                    log.warn(String.format(
                        "Exception caught from protectedRun of %s with origin '%s' in '%s'. Stopping processing",
                        sourcePayload, sourcePayload.getData(Payload.ORIGIN), this),
                             e);

                    // We don't close in a 'finally' clause because we shouldn't
                    // clean up if the JVM raises an Error type throwable<commit_msg>Logging: Better log for ThreadedStreamParser<commit_after>/*
 *  Licensed under the Apache License, Version 2.0 (the "License");
 *  you may not use this file except in compliance with the License.
 *  You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 *  Unless required by applicable law or agreed to in writing, software
 *  distributed under the License is distributed on an "AS IS" BASIS,
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  See the License for the specific language governing permissions and
 *  limitations under the License.
 *
 */
package dk.statsbiblioteket.summa.ingest.split;

import dk.statsbiblioteket.summa.common.Logging;
import dk.statsbiblioteket.summa.common.Record;
import dk.statsbiblioteket.summa.common.configuration.Configuration;
import dk.statsbiblioteket.summa.common.filter.Payload;
import dk.statsbiblioteket.summa.common.filter.PayloadQueue;
import dk.statsbiblioteket.util.qa.QAInfo;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import java.util.NoSuchElementException;
import java.util.concurrent.TimeUnit;

/**
 * Helper implementation of StreamParser that handles the bookkeeping of
 * threaded parsing. With threaded parsing, a thread is responsible for reading
 * from the Stream, producing Records and adding the Records to a queue.
 * Reading from the queue is done from outside the Thread.
 * </p><p>
 * Implementators of this abstract class only needs to override the
 * {@link #protectedRun(Payload)} method. They should check {@link #running}
 * frequently and stop processing if it is false.
 */
@QAInfo(level = QAInfo.Level.NORMAL,
        state = QAInfo.State.IN_DEVELOPMENT,
        author = "te")
public abstract class ThreadedStreamParser implements StreamParser {
    private static Log log = LogFactory.getLog(ThreadedStreamParser.class);

    /**
     * The maximum queue size, counted in number of Records.
     * </p><p>
     * Optional. Default is 100.
     */
    public static final String CONF_QUEUE_SIZE = "summa.ingest.stream.threadedstreamparser.queue.size";
    public static final int DEFAULT_QUEUE_SIZE = 100;

    /**
     * The maximum queue size, counted in bytes.
     * </p><p>
     * Optional. Default is 5 MB.
     */
    public static final String CONF_QUEUE_BYTESIZE = "summa.ingest.stream.threadedstreamparser.queue.bytesize";
    public static final int DEFAULT_QUEUE_BYTESIZE = 5*1000*1000; // 5 MB

    /**
     * The maximum number of milliseconds to wait for data when hasNext(),
     * next() or pump() is called.
     * </p><p>
     * Optional. Default is Integer.MAX_VALUE.
     */
    public static final String CONF_QUEUE_TIMEOUT = "summa.ingest.stream.threadedstreamparser.queue.timeout";
    public static final int DEFAULT_QUEUE_TIMEOUT = Integer.MAX_VALUE;
    private int queueTimeout = DEFAULT_QUEUE_TIMEOUT;

    //private static final long HASNEXT_SLEEP = 50; // Sleep-ms between polls
    private static final Payload INTERRUPTOR = new Payload(new Record("dummyID", "dummyStreamBase", new byte[0]));

    /*
     * Holds the produced Payloads. Generated Payloads will always be followed
     * by {@link #INTERRUPTOR}.
     */
    private PayloadQueue queue;
    /*
     * The next Payload to deliver. This is assigned by {@link #hasNext} and
     * used by {@link #next}. toDeliver will never be {@link INTERRUPTOR}.
     */
    private Payload toDeliver = null;

    /**
     * The source Payload that is to be processed.
     */
    private Payload sourcePayload;

    /**
     * If true, processing should proceed normally. If false, processing should
     * be terminated when it can be done without affecting stability.
     */
    protected boolean running = false;

    /**
     * empty == true iff no Payload has been assigned yet or
     * (the processing thread is not running and the
     * interruptor-Payload has been encountered in the queue).
     * </p><p>
     * {@link #open} must only be called when empty == true.
     */
    private boolean empty = true; // Starting condition

    private Throwable lastError = null;
    private Thread runningThread = null;
    private long queueCount = 0;

    public ThreadedStreamParser(Configuration conf) {
        queue = new PayloadQueue(conf.getInt(CONF_QUEUE_SIZE, DEFAULT_QUEUE_SIZE),
                                 conf.getInt(CONF_QUEUE_BYTESIZE, DEFAULT_QUEUE_BYTESIZE));
        queueTimeout = conf.getInt(CONF_QUEUE_TIMEOUT, queueTimeout);
        log.debug("Constructed ThreadedStreamParser with queue-size " + queue.remainingCapacity()
                  + " and queue timeout " + queueTimeout + " ms");
    }

    @Override
    public void open(Payload streamPayload) {
        //noinspection DuplicateStringLiteralInspection
        log.debug("open(" + streamPayload + ") called");
        if (!empty) {
            throw new IllegalStateException(String.format(
                    "Already parsing %s when open(%s) was called",
                    sourcePayload, streamPayload));
        }
        queue.clear(); // Clean-up from previous runs
        if (streamPayload.getStream() == null) {
            log.warn("No stream in received " + streamPayload + ". No Records will be generated");
            empty = true;
            return;
        }
        sourcePayload = streamPayload;
        startThread();
    }

    private void startThread() {
        log.trace("Starting Thread for " + sourcePayload);
        running = true;
        empty = false;
        /* Set up a thread reporting errors back to us */
        //noinspection DuplicateStringLiteralInspection
        runningThread = new Thread(new Runnable() {
            @SuppressWarnings("ObjectToString")
            @Override
            public void run() {
                log.trace("run() entered");
                try {
                    protectedRun(sourcePayload);
                    if (autoClose()) {
                        sourcePayload.close();
                    }
                } catch (Exception e) {
                    // TODO: Introduce option that fails ingest on this
                    String message = String.format("Exception caught from protectedRun of %s with origin '%s'",
                                                   sourcePayload, sourcePayload.getData(Payload.ORIGIN));
                    log.warn(String.format("%s in '%s'. Stopping processing", message, this), e);
                    Logging.logProcess("ThreadedStreamParser", message, Logging.LogLevel.WARN, sourcePayload);
                    // We don't close in a 'finally' clause because we shouldn't
                    // clean up if the JVM raises an Error type throwable
{'fraction_non_alphanumeric': 0.08865463370241165, 'fraction_numerical': 0.0037919005005308663, 'mean_word_length': 2.5863475659505033, 'pattern_counts': {'":': 0, '<': 23, '<?xml version=': 0, '>': 23, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2374382', 'n_tokens_mistral': 3568, 'n_tokens_neox': 3443, 'n_words': 1168}