<commit_before>package com.widowcrawler.core.queue;

import com.amazonaws.services.sqs.AmazonSQSAsyncClient;
import com.amazonaws.services.sqs.model.QueueDoesNotExistException;
import com.amazonaws.services.sqs.model.SendMessageRequest;
import com.amazonaws.services.sqs.model.SendMessageResult;
import com.netflix.governator.annotations.AutoBindSingleton;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.Validate;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import javax.inject.Inject;
import java.util.*;
import java.util.concurrent.*;

/**
 * @author <NAME>
 */
@AutoBindSingleton
public class QueueManager {

    private final Logger logger = LoggerFactory.getLogger(getClass());

    @Inject
    private AmazonSQSAsyncClient sqsClient;

    private Map<String, ConcurrentLinkedQueue<Message>> messagesMap = null;
    private Map<String, String> queueUrls = null;

    private ExecutorService executorService = null;
    private Queue<Future<SendMessageResult>> enqueueActions = null;

    private Runnable enququer = () -> {

        // TODO: Impose some sort of limit on the number of retries per message

        //noinspection InfiniteLoopStatement
        while (true) {
            Future<SendMessageResult> future = null;

            while (future == null) {
                future = enqueueActions.poll();

                if (future == null) {
                    Thread.yield();
                }
            }

            try {
                SendMessageResult result = future.get();
                logger.info("Message enqueued successfully. Message ID: " + result.getMessageId());

            } catch (InterruptedException e) {
                logger.error("Enqueuer interrupted", e);
                Thread.currentThread().interrupt();

            } catch (ExecutionException e) {
                // eat the error, because wtf else
                logger.error("Enqueueing failed", e.getCause());
            }
        }
    };

    private Runnable messagePoller = () -> {
        // if < 10 messages already in queue && no in-progress fetch, fetch
        // else no

        //noinspection InfiniteLoopStatement
        while (true) {
            for (Map.Entry<String, String> entry : queueUrls.entrySet()) {
                String queueName = entry.getKey();
                String queueUrl = entry.getValue();

                if (messagesMap.get(queueName).size() < 1) {
                    // pull new messages
                    List<com.amazonaws.services.sqs.model.Message> messages = sqsClient.receiveMessage(queueUrl).getMessages();

                    // Convert Amazon SQS message to our general message type
                    for (com.amazonaws.services.sqs.model.Message message : messages) {
                        Message genericMessage = new Message(message.getBody(), message.getMessageId(), message.getReceiptHandle());
                        messagesMap.get(queueName).offer(genericMessage);
                    }
                }
            }

            Thread.yield();
        }
    };

    public QueueManager() { }
<commit_msg>Enabling batch-get and long-polling for sqs<commit_after>package com.widowcrawler.core.queue;

import com.amazonaws.services.sqs.AmazonSQSAsyncClient;
import com.amazonaws.services.sqs.model.QueueDoesNotExistException;
import com.amazonaws.services.sqs.model.ReceiveMessageRequest;
import com.amazonaws.services.sqs.model.SendMessageRequest;
import com.amazonaws.services.sqs.model.SendMessageResult;
import com.netflix.governator.annotations.AutoBindSingleton;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.Validate;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import javax.inject.Inject;
import java.util.*;
import java.util.concurrent.*;

/**
 * @author <NAME>
 */
@AutoBindSingleton
public class QueueManager {

    private final Logger logger = LoggerFactory.getLogger(getClass());

    @Inject
    private AmazonSQSAsyncClient sqsClient;

    private Map<String, ConcurrentLinkedQueue<Message>> messagesMap = null;
    private Map<String, String> queueUrls = null;

    private ExecutorService executorService = null;
    private Queue<Future<SendMessageResult>> enqueueActions = null;

    private Runnable enququer = () -> {

        // TODO: Impose some sort of limit on the number of retries per message
        // dead-letter queue for sure, probably through AWS config

        //noinspection InfiniteLoopStatement
        while (true) {
            Future<SendMessageResult> future = null;

            while (future == null) {
                future = enqueueActions.poll();

                if (future == null) {
                    Thread.yield();
                }
            }

            try {
                SendMessageResult result = future.get();
                logger.info("Message enqueued successfully. Message ID: " + result.getMessageId());

            } catch (InterruptedException e) {
                logger.error("Enqueuer interrupted", e);
                Thread.currentThread().interrupt();

            } catch (ExecutionException e) {
                // eat the error, because wtf else
                logger.error("Enqueueing failed", e.getCause());
            }
        }
    };

    private Runnable messagePoller = () -> {
        // if < 10 messages already in queue && no in-progress fetch, fetch
        // else no

        //noinspection InfiniteLoopStatement
        while (true) {
            for (Map.Entry<String, String> entry : queueUrls.entrySet()) {
                String queueName = entry.getKey();
                String queueUrl = entry.getValue();

                if (messagesMap.get(queueName).size() < 1) {
                    // pull new messages in batches with long-polling enabled
                    ReceiveMessageRequest rmr = new ReceiveMessageRequest(queueUrl)
                            .withMaxNumberOfMessages(10)
                            .withWaitTimeSeconds(10);
                    List<com.amazonaws.services.sqs.model.Message> messages = sqsClient.receiveMessage(rmr).getMessages();

                    // Convert Amazon SQS message to our general message type
                    for (com.amazonaws.services.sqs.model.Message message : messages) {
                        Message genericMessage = new Message(message.getBody(), message.getMessageId(), message.getReceiptHandle());
                        messagesMap.get(queueName).offer(genericMessage);
                    }
                }
            }

            Thread.yield();
        }
    };

    public QueueManager() { }

{'fraction_non_alphanumeric': 0.09751434034416825, 'fraction_numerical': 0.002647448154140315, 'mean_word_length': 2.216650898770104, 'pattern_counts': {'":': 0, '<': 25, '<?xml version=': 0, '>': 25, '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': '17799', 'n_tokens_mistral': 1792, 'n_tokens_neox': 1704, 'n_words': 398}