<commit_before>package com.github.ivan_osipov.clabo.api.internal

import com.github.ivan_osipov.clabo.api.output.dto.SendParams
import com.google.common.collect.Queues
import org.slf4j.Logger
import org.slf4j.LoggerFactory
import java.util.*
import java.util.concurrent.Executor
import java.util.concurrent.Executors
import java.util.concurrent.Semaphore

internal class QueueBasedSender(val apiInteraction: TelegramApiInteraction) {

    private val outputQueue: Queue<SendParams> = Queues.newConcurrentLinkedQueue()

    private val queueSemaphore: Semaphore = Semaphore(0)

    private val resultTakingSemaphore: Semaphore = Semaphore(1)

    private val workerExecutor: Executor = Executors.newSingleThreadExecutor()

    private val logger: Logger = LoggerFactory.getLogger(QueueBasedSender::class.java)

    init {
        workerExecutor.execute(WorkerThread())
        logger.debug("QueueBasedSender is initialized")
    }

    fun send(sendParams: SendParams) {
        outputQueue.add(sendParams)
        logger.debug("New message in queue")
        queueSemaphore.release()
    }

    inner class WorkerThread : Runnable {

        private val logger: Logger = LoggerFactory.getLogger(WorkerThread::class.java)

        override fun run() {
            while (true) {
                val nextOutputMessage: SendParams? = outputQueue.poll()

                if(nextOutputMessage == null) {
                    logger.debug("Output queue is empty")
                    queueSemaphore.acquire()
                } else {
                    logger.debug("Message sending")
                    resultTakingSemaphore.acquire()

                    apiInteraction.sendMessage(nextOutputMessage, {
                        resultTakingSemaphore.release()
                    }, {
                        resultTakingSemaphore.release()
                    })
                }
            }
        }
    }
}<commit_msg>updated a sender sync by chats<commit_after>package com.github.ivan_osipov.clabo.api.internal

import com.github.ivan_osipov.clabo.api.output.dto.SendParams
import com.github.ivan_osipov.clabo.utils.ChatId
import com.google.common.base.Preconditions.checkArgument
import com.google.common.collect.Maps
import com.google.common.collect.Queues
import org.slf4j.Logger
import org.slf4j.LoggerFactory
import java.util.*
import java.util.concurrent.Executor
import java.util.concurrent.Executors
import java.util.concurrent.Semaphore
import java.util.concurrent.atomic.AtomicBoolean

internal class QueueBasedSender(val apiInteraction: TelegramApiInteraction) {

    val messageProcessingSemaphore = Semaphore(0)

    private val chatOutputMap: MutableMap<ChatId, ChatInfo> = Maps.newHashMap()

    private val workerExecutor: Executor = Executors.newSingleThreadExecutor() //executes thread for queue processing

    private val logger: Logger = LoggerFactory.getLogger(QueueBasedSender::class.java)

    init {
        workerExecutor.execute(WorkerThread())
        logger.debug("QueueBasedSender is initialized")
    }

    fun send(message: SendParams) {
        var chatOutput = chatOutputMap[message.chatId]
        if (chatOutput == null) {
            chatOutput = ChatInfo(message.chatId)
            chatOutputMap[message.chatId] = chatOutput
        }
        chatOutput.outputQueue.add(message)
        logger.debug("New message in queue (chatId: ${chatOutput.chatId})")
        notifyWorker()
    }

    inner class WorkerThread : Runnable {

        private val logger: Logger = LoggerFactory.getLogger(WorkerThread::class.java)

        override fun run() {
            while (true) {
                val chatsForProcessing = chatOutputMap.filter { it.value.canBeProcessed() }.values
                if(chatsForProcessing.isEmpty()) {
                    messageProcessingSemaphore.acquire()
                    logger.debug("The worker thread is unlocked")
                    continue
                }
                for (chatForProcessing in chatsForProcessing) {

                    val message = chatForProcessing.outputQueue.peek()
                    logger.debug("Message sending")

                    val lockIsSuccessful = chatForProcessing.resultTalkingMutex.compareAndSet(false, true)

                    if (lockIsSuccessful) {
                        val headMessage = chatForProcessing.outputQueue.remove()
                        checkArgument(headMessage == message, "Incorrect message as the queue head")

                        apiInteraction.sendMessage(message, {
                            chatForProcessing.resultTalkingMutex.set(false)
                            notifyWorker()
                        }, {
                            chatForProcessing.resultTalkingMutex.set(false)
                            notifyWorker()
                        })
                    }
                }
            }
        }
    }

    private fun notifyWorker() {
        if (messageProcessingSemaphore.hasQueuedThreads()) {
            messageProcessingSemaphore.release()
        }
    }

    private class ChatInfo(val chatId: ChatId,
                   val resultTalkingMutex: AtomicBoolean = AtomicBoolean(false),
                   val outputQueue: Queue<SendParams> = Queues.newLinkedBlockingQueue()) {
        fun canBeProcessed() = !resultTalkingMutex.get() && outputQueue.isNotEmpty()
    }
}
{'fraction_non_alphanumeric': 0.08230683090705487, 'fraction_numerical': 0.0013064576334453153, 'mean_word_length': 2.283700980392157, 'pattern_counts': {'":': 0, '<': 6, '<?xml version=': 0, '>': 6, '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': '6990260', 'n_tokens_mistral': 1421, 'n_tokens_neox': 1345, 'n_words': 261}