<commit_before>package funnel

import java.net.URI

import com.twitter.algebird.Group
import org.scalacheck.Arbitrary._
import org.scalacheck.Prop._
import org.scalacheck._
import java.util.concurrent.atomic.AtomicBoolean
import java.util.concurrent.Executors

import scala.concurrent.duration._
import scalaz.Nondeterminism
import scalaz.concurrent.{Strategy, Task}
import scalaz.std.list._
import scalaz.std.tuple._
import scalaz.stream.Cause.{End, Kill, Terminated}
import scalaz.stream.Process.Halt
import scalaz.stream.{Process, Sink}
import scalaz.syntax.foldable._
import scalaz.syntax.functor._

object ProcessMirroringEventsSpec extends Properties("processMirroringEvents") {

  implicit val s = Executors.newScheduledThreadPool(4)
  private val clusterName: ClusterName = "clusterName"

  property("keys are being removed upon disconnect") = secure {
    val M = Monitoring.instance(Monitoring.defaultPool, println)
    val i = new Instruments(monitoring = M)
    val gauge = i.gauge("someGauge",0.0)
    val key = gauge.key

    //closing the signal, simulating the host dropping the connection
    M.get(key).close.run

    val x = M.keySenescence(Events.every(1 milliseconds),Process.eval(Task.delay(gauge.key)))

    Thread.sleep(1000)

    x.run.run
    !M.keys.get.run.contains(key)
  }


  property("TCP disconnect removes host from /mirror/sources list") = secure {
    val M = Monitoring.instance(Monitoring.defaultPool, println)
    val enqueueSink: Sink[Task, Command] = M.mirroringQueue.enqueue
    val uri= URI.create("http://localhost")
    val mirror = Mirror(uri,clusterName)
    val discard = Discard(uri)
    val datapoint = new Datapoint[Any](Key[String]("key",Units.TrafficLight),"green")
    val commands: Process[Task, Command] = Process.emitAll(Seq(mirror, discard))

    //this should be commands.to(enqueueSink) but scala 2.10 didn't like it
    //val commandEnqueue = commands.to(enqueueSink)
    val commandEnqueue = commands.zipWith(enqueueSink)((o,f) => f(o)).eval

    val mockParse: URI => Process[Task, Datapoint[Any]] = 
      _ => Process.eval(Task.delay(throw new RuntimeException("boom")))

    //enqueue the commands
    commandEnqueue.run.run

    //start processing commands
    Task.fork(
      M.processMirroringEvents(
        mockParse,
        nodeRetries = _ => Events.takeEvery(1 millisecond,1)
      )
    ).runAsync(_ => ())

    //Let it run
    Thread.sleep(1000)

    //cleanup
    M.mirroringQueue.close.run<commit_msg>Fixing compile error due to postfix operation<commit_after>package funnel

import java.net.URI

import com.twitter.algebird.Group
import org.scalacheck.Arbitrary._
import org.scalacheck.Prop._
import org.scalacheck._
import java.util.concurrent.atomic.AtomicBoolean
import java.util.concurrent.Executors

import scala.concurrent.duration._
import scalaz.Nondeterminism
import scalaz.concurrent.{Strategy, Task}
import scalaz.std.list._
import scalaz.std.tuple._
import scalaz.stream.Cause.{End, Kill, Terminated}
import scalaz.stream.Process.Halt
import scalaz.stream.{Process, Sink}
import scalaz.syntax.foldable._
import scalaz.syntax.functor._

object ProcessMirroringEventsSpec extends Properties("processMirroringEvents") {

  implicit val s = Executors.newScheduledThreadPool(4)
  private val clusterName: ClusterName = "clusterName"

  property("keys are being removed upon disconnect") = secure {
    val M = Monitoring.instance(Monitoring.defaultPool, println)
    val i = new Instruments(monitoring = M)
    val gauge = i.gauge("someGauge",0.0)
    val key = gauge.key

    //closing the signal, simulating the host dropping the connection
    M.get(key).close.run

    val x = M.keySenescence(Events.every(1.milliseconds),Process.eval(Task.delay(gauge.key)))

    Thread.sleep(1000)

    x.run.run
    !M.keys.get.run.contains(key)
  }


  property("TCP disconnect removes host from /mirror/sources list") = secure {
    val M = Monitoring.instance(Monitoring.defaultPool, println)
    val enqueueSink: Sink[Task, Command] = M.mirroringQueue.enqueue
    val uri= URI.create("http://localhost")
    val mirror = Mirror(uri,clusterName)
    val discard = Discard(uri)
    val datapoint = new Datapoint[Any](Key[String]("key",Units.TrafficLight),"green")
    val commands: Process[Task, Command] = Process.emitAll(Seq(mirror, discard))

    //this should be commands.to(enqueueSink) but scala 2.10 didn't like it
    //val commandEnqueue = commands.to(enqueueSink)
    val commandEnqueue = commands.zipWith(enqueueSink)((o,f) => f(o)).eval

    val mockParse: URI => Process[Task, Datapoint[Any]] = 
      _ => Process.eval(Task.delay(throw new RuntimeException("boom")))

    //enqueue the commands
    commandEnqueue.run.run

    //start processing commands
    Task.fork(
      M.processMirroringEvents(
        mockParse,
        nodeRetries = _ => Events.takeEvery(1.millisecond,1)
      )
    ).runAsync(_ => ())

    //Let it run
    Thread.sleep(1000)

    //cleanup
    M.mirroringQueue.close.run
{'fraction_non_alphanumeric': 0.11373648378053665, 'fraction_numerical': 0.006808169803764518, 'mean_word_length': 3.99000999000999, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 13, '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': '5895429', 'n_tokens_mistral': 1629, 'n_tokens_neox': 1549, 'n_words': 349}