<commit_before>package com.sksamuel.akka.patterns

import akka.actor.{ActorRef, Actor}
import scala.collection.mutable.ListBuffer

/**
 * Holds up further messages until the previous messages have
 * been acknowledged.
 *
 * @author <NAME> */
class FlowControlActor(target: ActorRef, windowSize: Int = 1) extends Actor {

  val buffer = new ListBuffer[AnyRef]
  val pending = 0

  def receive = {
    case Acknowledged =>
      //blocked = false
      if (buffer.size > 0) send(buffer.remove(0))
    case msg: AnyRef =>
    //  if (blocked) buffer append msg
    //else send(msg)
  }

  def send(msg: AnyRef) = {
    target ! msg
    // blocked = true
  }
}<commit_msg>Updated flow control actor for window size<commit_after>package com.sksamuel.akka.patterns

import akka.actor.{ActorRef, Actor}
import scala.collection.mutable

/**
 * Holds up further messages until the previous messages have
 * been acknowledged.
 *
 * @author <NAME> */
class FlowControlActor(target: ActorRef, windowSize: Int = 1) extends Actor {

  val queue = mutable.Queue.empty[Any]
  var pending = 0

  def receive = {
    case Acknowledged =>
      if (pending > 0) pending = pending - 1
      if (queue.size > 0) {
        target ! queue.dequeue()
        pending = pending + 1
      }
    case msg =>
      if (pending == windowSize) {
        queue enqueue msg
      } else {
        pending = pending + 1
        target ! msg
      }
  }
}
{'fraction_non_alphanumeric': 0.10689170182841069, 'fraction_numerical': 0.007735583684950774, 'mean_word_length': 2.696103896103896, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 12, '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': '2142718', 'n_tokens_mistral': 456, 'n_tokens_neox': 435, 'n_words': 134}