<commit_before>package eu.liveandgov.wp1.pipeline.impl;

import eu.liveandgov.wp1.data.CallbackSet;
import eu.liveandgov.wp1.data.Stoppable;
import eu.liveandgov.wp1.pipeline.Pipeline;

import java.util.concurrent.*;

/**
 * Created by <NAME> on 01.03.14.
 */
public class Player<Item> extends Pipeline<Item, Item> implements Stoppable {
    /**
     * The service executing the playback
     */
    public final ScheduledExecutorService scheduledExecutorService;

    /**
     * The limit of items to store in the queue or -1 for no limit
     */
    public final int limit;

    /**
     * The limit of items simultaneously played back or -1 to play all queued
     */
    public final int playbackLimit;

    /**
     * The rate of playback
     */
    public final long playbackRate;

    /**
     * Event called with false when playback is suspended
     */
    public final CallbackSet<Boolean> suspended = new CallbackSet<Boolean>();

    /**
     * Event called with true when playback is resumed
     */
    public final CallbackSet<Boolean> resumed = new CallbackSet<Boolean>();

    /**
     * All items of this player
     */
    private final ConcurrentLinkedQueue<Item> queue;

    /**
     * The task handling the playback
     */
    private ScheduledFuture<?> task;

    /**
     * Creates a new player with the given parameters
     *
     * @param scheduledExecutorService The executor service used to schedule the playback
     * @param limit                    The limit of items to store in the queue or -1 for no limit
     * @param playbackLimit            The limit of items simultaneously played back or -1 to play all queued
     * @param playbackRate             The rate of playback
     */
    public Player(ScheduledExecutorService scheduledExecutorService, int limit, int playbackLimit, long playbackRate) {
        this.scheduledExecutorService = scheduledExecutorService;
        this.limit = limit;
        this.playbackLimit = playbackLimit;
        this.playbackRate = playbackRate;

        queue = new ConcurrentLinkedQueue<Item>();
    }

    private void resume() {
        task = scheduledExecutorService.scheduleWithFixedDelay(playerJob, 0L, playbackRate, TimeUnit.MILLISECONDS);

        resumed.call(true);
    }

    private void suspend() {
        task.cancel(true);
        task = null;

        suspended.call(false);
    }

    @Override
    public void push(final Item item) {
        if (queue.size() + 1 <= limit || limit == -1) {
            queue.offer(item);
        }

        if (task == null) {
            resume();
        }
    }

    private final Runnable playerJob = new Runnable() {
        @Override
        public void run() {
            for (int i = 0; !queue.isEmpty() && (i < playbackLimit || playbackLimit == -1); i++) {
                produce(queue.poll());
            }

            // If the queue is empty, stop this task if the queue has no more elements
            if (queue.isEmpty()) {
                suspend();
            }
        }
    };


    @Override
    public void stop() {
        suspend();
    }
}<commit_msg>Replaced scheduling and suspend/resume pattern with a blocking queue pattern similar to the connector thread and sensor queue approach<commit_after>package eu.liveandgov.wp1.pipeline.impl;

import eu.liveandgov.wp1.data.CallbackSet;
import eu.liveandgov.wp1.data.Stoppable;
import eu.liveandgov.wp1.pipeline.Pipeline;

import java.util.concurrent.*;

/**
 * Created by <NAME> on 01.03.14.
 */
public class Player<Item> extends Pipeline<Item, Item> implements Stoppable {
    /**
     * The service executing the playback
     */
    public final ExecutorService executorService;

    /**
     * The limit of items to store in the queue or -1 for no limit
     */
    public final int limit;

    /**
     * All items of this player
     */
    private final BlockingQueue<Item> queue;

    /**
     * The task handling the playback
     */
    private final Future<?> task;

    /**
     * Creates a new player with the given parameters
     *
     * @param executorService The executor service used to run the playback
     * @param limit           The limit of items to store in the queue or -1 for no limit
     */
    public Player(ExecutorService executorService, int limit) {
        this.executorService = executorService;
        this.limit = limit;

        queue = new LinkedBlockingQueue<Item>();
        task = executorService.submit(playerJob);
    }

    @Override
    public void push(final Item item) {
        if (limit == -1 || queue.size() < limit)
            try {
                queue.put(item);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
    }

    private final Runnable playerJob = new Runnable() {
        @Override
        public void run() {
            while (true) {
                try {
                    produce(queue.take());
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }
        }
    };


    @Override
    public void stop() {
        task.cancel(true);
    }
}
{'fraction_non_alphanumeric': 0.09300077339520495, 'fraction_numerical': 0.006187161639597835, 'mean_word_length': 2.0957510472770795, 'pattern_counts': {'":': 0, '<': 22, '<?xml version=': 0, '>': 19, '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': '2147728', 'n_tokens_mistral': 1420, 'n_tokens_neox': 1351, 'n_words': 446}