<commit_before>package com.twitter.finagle.kestrel.unit

import _root_.java.net.{InetSocketAddress, SocketAddress}
import _root_.java.nio.charset.Charset
import _root_.java.util.concurrent._
import com.google.common.cache.{LoadingCache, CacheLoader, CacheBuilder}
import com.twitter.concurrent.{Spool, Broker}
import com.twitter.conversions.time._
import com.twitter.finagle.builder.StaticCluster
import com.twitter.finagle.builder.{ClientConfig, ClientBuilder, Cluster}
import com.twitter.finagle.kestrel._
import com.twitter.finagle.kestrel.protocol._
import com.twitter.finagle.memcached.util.ChannelBufferUtils._
import com.twitter.finagle.{ClientConnection, Service, ServiceFactory}
import com.twitter.util.{Await, Future, Promise, Return, Time}
import org.jboss.netty.buffer.ChannelBuffer
import org.specs.SpecificationWithJUnit
import org.specs.mock.Mockito
import scala.collection.mutable.{ArrayBuffer, Set => MSet}

class MultiReaderSpec extends SpecificationWithJUnit with Mockito {
  noDetailedDiffs()

  class MockHandle extends ReadHandle {
    val _messages = new Broker[ReadMessage]
    val _error = new Broker[Throwable]

    val messages = _messages.recv
    val error = _error.recv
    def close() {} // to spy on!
  }

  "MultiReader" should {
    "old-style, static ReadHandle cluster" in {
      val N = 3
      val handles = (0 until N).toSeq map { _ => spy(new MockHandle) }
      val cluster: Cluster[ReadHandle] = new StaticCluster[ReadHandle](handles)

      "always grab the first available message" in {
        val handle = MultiReader.merge(cluster)

        val messages = new ArrayBuffer[ReadMessage]
        handle.messages foreach { messages += _ }

        // stripe some messages across
        val sentMessages = 0 until N*100 map { _ => mock[ReadMessage] }

        messages must beEmpty
        sentMessages.zipWithIndex foreach { case (m, i) =>
          handles(i % handles.size)._messages ! m
        }

        messages must be_==(sentMessages)
      }

      // We use frozen time for deterministic randomness.
      // The message output order was simply determined empirically.
      "round robin from multiple available queues" in Time.withTimeAt(Time.epoch + 1.seconds) { _ =>
        // stuff the queues beforehand
        val ms = handles map { h =>
          val m = mock[ReadMessage]
          h._messages ! m
          m
        }

        val handle = MultiReader.merge(cluster)
        (handle.messages??) must be_==(ms(0))
        (handle.messages??) must be_==(ms(2))
        (handle.messages??) must be_==(ms(1))
      }

      "propagate closes" in {
        handles foreach { h => there was no(h).close() }<commit_msg>[split] Tagging MultiReader test as flaky<commit_after>package com.twitter.finagle.kestrel.unit

import _root_.java.net.{InetSocketAddress, SocketAddress}
import _root_.java.nio.charset.Charset
import _root_.java.util.concurrent._
import com.google.common.cache.{LoadingCache, CacheLoader, CacheBuilder}
import com.twitter.concurrent.{Spool, Broker}
import com.twitter.conversions.time._
import com.twitter.finagle.builder.StaticCluster
import com.twitter.finagle.builder.{ClientConfig, ClientBuilder, Cluster}
import com.twitter.finagle.kestrel._
import com.twitter.finagle.kestrel.protocol._
import com.twitter.finagle.memcached.util.ChannelBufferUtils._
import com.twitter.finagle.{ClientConnection, Service, ServiceFactory}
import com.twitter.util.{Await, Future, Promise, Return, Time}
import org.jboss.netty.buffer.ChannelBuffer
import org.specs.SpecificationWithJUnit
import org.specs.mock.Mockito
import scala.collection.mutable.{ArrayBuffer, Set => MSet}

class MultiReaderSpec extends SpecificationWithJUnit with Mockito {
  noDetailedDiffs()

  class MockHandle extends ReadHandle {
    val _messages = new Broker[ReadMessage]
    val _error = new Broker[Throwable]

    val messages = _messages.recv
    val error = _error.recv
    def close() {} // to spy on!
  }

  "MultiReader" should {
    "old-style, static ReadHandle cluster" in {
      val N = 3
      val handles = (0 until N).toSeq map { _ => spy(new MockHandle) }
      val cluster: Cluster[ReadHandle] = new StaticCluster[ReadHandle](handles)

      "always grab the first available message" in {
        val handle = MultiReader.merge(cluster)

        val messages = new ArrayBuffer[ReadMessage]
        handle.messages foreach { messages += _ }

        // stripe some messages across
        val sentMessages = 0 until N*100 map { _ => mock[ReadMessage] }

        messages must beEmpty
        sentMessages.zipWithIndex foreach { case (m, i) =>
          handles(i % handles.size)._messages ! m
        }

        messages must be_==(sentMessages)
      }

      // We use frozen time for deterministic randomness.
      // The message output order was simply determined empirically.
      if (!sys.props.contains("SKIP_FLAKY")) {
        "round robin from multiple available queues" in Time.withTimeAt(Time.epoch + 1.seconds) { _ =>
          // stuff the queues beforehand
          val ms = handles map { h =>
            val m = mock[ReadMessage]
            h._messages ! m
            m
          }

          val handle = MultiReader.merge(cluster)
          (handle.messages??) must be_==(ms(0))
          (handle.messages??) must be_==(ms(2))
          (handle.messages??) must be_==(ms(1))
        }
      }

      "propagate closes" in {
        handles foreach { h => there was no(h).close() }
{'fraction_non_alphanumeric': 0.1015925315760571, 'fraction_numerical': 0.003660992128866923, 'mean_word_length': 3.120663650075415, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 17, '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': '6003293', 'n_tokens_mistral': 1610, 'n_tokens_neox': 1560, 'n_words': 417}