<commit_before>package com.twitter.gizzard.scheduler

import java.util.{LinkedList => JLinkedList}
import java.nio.ByteBuffer
import scala.collection.mutable.Queue
import scala.util.matching.Regex
import com.twitter.rpcclient.LoadBalancingChannel
import com.twitter.util.Duration
import thrift.{JobInjector, JobInjectorClient}
import thrift.conversions.Sequences._
import nameserver.JobRelay


class ReplicatingJsonCodec(relay: => JobRelay, unparsable: Array[Byte] => Unit)
extends JsonCodec[JsonJob](unparsable) {
  lazy val innerCodec = {
    val c = new JsonCodec[JsonJob](unparsable)
    c += ("RemoteReplicatingJob".r -> new RemoteReplicatingJobParser(c, relay))
    c
  }

  override def +=(item: (Regex, JsonJobParser[JsonJob])) = innerCodec += item
  override def +=(r: Regex, p: JsonJobParser[JsonJob])   = innerCodec += ((r, p))

  override def inflate(json: Map[String, Any]): JsonJob = {
    innerCodec.inflate(json) match {
      case j: RemoteReplicatingJob[_] => j
      case j => if (j.shouldReplicate) {
        new RemoteReplicatingJob(relay, List(j))
      } else {
        j
      }
    }
  }
}

class RemoteReplicatingJob[J <: JsonJob](
  relay: JobRelay,
  jobs: Iterable[J],
  destClusters: Iterable[String])
extends JsonNestedJob(jobs) {
  def this(relay: JobRelay, jobs: Iterable[J]) = this(relay, jobs, relay.clusters)

  val clustersQueue = {
    val q = new Queue[String]
    q ++= destClusters
    q
  }

  override def toMap: Map[String, Any] =
    Map("dest_clusters" -> clustersQueue.toList :: super.toMap.toList: _*)

  // XXX: do all this work in parallel in a future pool.
  override def apply() {
    while (!clustersQueue.isEmpty) {
      val c = clustersQueue.dequeue()
      try { relay(c)(List(this)) } catch {
        case e: Throwable => clustersQueue += c; throw e
      }
    }

    super.apply()
  }
}

class RemoteReplicatingJobParser[J <: JsonJob](
  codec: JsonCodec[J],
  relay: => JobRelay)
extends JsonJobParser[J] {
  type TaskJsons = Iterable[Map[String, Any]]
<commit_msg>make sure we throw exceptions from failed replication<commit_after>package com.twitter.gizzard.scheduler

import java.util.{LinkedList => JLinkedList}
import java.nio.ByteBuffer
import scala.collection.mutable.Queue
import scala.util.matching.Regex
import com.twitter.rpcclient.LoadBalancingChannel
import com.twitter.util.Duration
import thrift.{JobInjector, JobInjectorClient}
import thrift.conversions.Sequences._
import nameserver.JobRelay
import net.lag.logging.Logger


class ReplicatingJsonCodec(relay: => JobRelay, unparsable: Array[Byte] => Unit)
extends JsonCodec[JsonJob](unparsable) {
  lazy val innerCodec = {
    val c = new JsonCodec[JsonJob](unparsable)
    c += ("RemoteReplicatingJob".r -> new RemoteReplicatingJobParser(c, relay))
    c
  }

  override def +=(item: (Regex, JsonJobParser[JsonJob])) = innerCodec += item
  override def +=(r: Regex, p: JsonJobParser[JsonJob])   = innerCodec += ((r, p))

  override def inflate(json: Map[String, Any]): JsonJob = {
    innerCodec.inflate(json) match {
      case j: RemoteReplicatingJob[_] => j
      case j => if (j.shouldReplicate) {
        new RemoteReplicatingJob(relay, List(j))
      } else {
        j
      }
    }
  }
}

class RemoteReplicatingJob[J <: JsonJob](
  relay: JobRelay,
  jobs: Iterable[J],
  destClusters: Iterable[String])
extends JsonNestedJob(jobs) {
  def this(relay: JobRelay, jobs: Iterable[J]) = this(relay, jobs, relay.clusters)

  val clustersQueue = {
    val q = new Queue[String]
    q ++= destClusters
    q
  }

  override def toMap: Map[String, Any] =
    Map("dest_clusters" -> clustersQueue.toList :: super.toMap.toList: _*)

  // XXX: do all this work in parallel in a future pool.
  override def apply() {
    while (!clustersQueue.isEmpty) {
      val c = clustersQueue.dequeue()
      try {
        relay(c)(List(this)) } catch {
        case e: Throwable => {
          clustersQueue += c
          throw e
        }
      }
    }

    super.apply()
  }
}

class RemoteReplicatingJobParser[J <: JsonJob](
  codec: JsonCodec[J],
  relay: => JobRelay)
extends JsonJobParser[J] {
  type TaskJsons = Iterable[Map[String, Any]]

{'fraction_non_alphanumeric': 0.12256669069935111, 'fraction_numerical': 0.0, 'mean_word_length': 3.692220969560316, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 21, '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': '7661733', 'n_tokens_mistral': 1397, 'n_tokens_neox': 1399, 'n_words': 327}