<commit_before>package funnel
package flask

import scala.concurrent.duration._
import scalaz.concurrent.{ Actor, Strategy, Task }
import scalaz.std.option._
import scalaz.syntax.applicative._
import org.scalatest.{FlatSpec,Matchers,BeforeAndAfterAll}
import scalaz.stream.Process
import scalaz.stream.async.signal
import com.amazonaws.auth.BasicAWSCredentials
import argonaut._, Argonaut._
import org.slf4j.LoggerFactory
import funnel.elastic._
import funnel.http.JSON._
import funnel.zeromq._
import sockets._

class FlaskSpec extends FlatSpec with Matchers with BeforeAndAfterAll {
  import java.io.File
  import knobs.{ ClassPathResource, Config, FileResource, Required }
  import dispatch._

  val config: Task[Config] = for {
    a <- knobs.loadImmutable(List(Required(
      FileResource(new File("/usr/share/oncue/etc/flask.cfg")) or
        ClassPathResource("oncue/flask.cfg"))))
    b <- knobs.aws.config
  } yield a ++ b

  val (options, cfg) = config.flatMap { cfg =>
    val name             = cfg.lookup[String]("flask.name").getOrElse("flask")
    val elasticURL       = cfg.lookup[String]("flask.elastic-search.url")
    val elasticIx        = cfg.lookup[String]("flask.elastic-search.index-name")
    val elasticTy        = cfg.lookup[String]("flask.elastic-search.type-name")
    val elasticDf        =
      cfg.lookup[String]("flask.elastic-search.partition-date-format").getOrElse("yyyy.MM.dd")
    val elasticTimeout   = cfg.lookup[Int]("flask.elastic-search.connection-timeout-in-ms").getOrElse(5000)
    val riemannHost      = cfg.lookup[String]("flask.riemann.host")
    val riemannPort      = cfg.lookup[Int]("flask.riemann.port")
    val ttl              = cfg.lookup[Int]("flask.riemann.ttl-in-minutes").map(_.minutes)
    val riemann          = (riemannHost |@| riemannPort |@| ttl)(RiemannCfg)
    val elastic          = (elasticURL |@| elasticIx |@| elasticTy)(
      ElasticCfg(_, _, _, elasticDf, elasticTimeout))
    val snsErrorTopic    = cfg.require[String]("flask.sns-error-topic")
    val awsAccessKey     = cfg.require[String]("aws.access-key")
    val awsSecretKey     = cfg.require[String]("aws.secret-key")
    val awsCredentials   = new BasicAWSCredentials(awsAccessKey, awsSecretKey)
    val awsProxyHost     = cfg.lookup[String]("aws.proxy-host")
    val awsProxyPort     = cfg.lookup[Int]("aws.proxy-port")
    val awsProxyProtocol = cfg.lookup[String]("aws.proxy-protocol")
    val awsRegion        = cfg.require[String]("aws.region")
    val port             = cfg.lookup[Int]("flask.network.port").getOrElse(5775)
    Task((Options(Option(name), elastic, riemann, snsErrorTopic, awsCredentials, awsProxyHost,
      awsProxyPort, awsProxyProtocol, awsRegion, port), cfg))
  }.run

  val logger = LoggerFactory.getLogger("flask")

  implicit val logPool: Strategy = Strategy.Executor(java.util.concurrent.Executors.newFixedThreadPool(1))

  val L = Actor.actor((s: String) => logger.info(s))

  implicit val log: String => Unit = s => L(s)

  val flaskUrl = url(s"http://localhost:${options.funnelPort}/mirror").setContentType("application/json", "UTF-8")

  val payload = s"""
  [
    {
      "bucket": "datapoints-1.0-us-east",
      "urls": [
        "${Settings.tcp}"
      ]
    }
  ]
  """

  val ready = signal[Boolean]
  ready.set(false)

  implicit val B = scalaz.std.anyVal.booleanInstance.conjunction
  implicit val s = scalaz.stream.DefaultScheduler

  val E = Endpoint.unsafeApply(publish &&& bind, Settings.tcp)

  implicit val batransport: Transportable[Array[Byte]] = Transportable { ba =>
    Transported(Schemes.unknown, Versions.v1, None, None, ba)
  }

  val seq: Seq[Array[Byte]] = for(i <- 0 until 5000) yield Datapoint(Key("now/life", Units.Count: Units[Double], "description", Map("url" -> "http://localhost")), 42.0).asJson.spaces2.getBytes
  val k: Seq[Boolean] = seq.map(_ => true) ++ Seq(false)

  val proc: Process[Task, Array[Byte]] = Process.emitAll(seq) fby Process.eval_(ready.set(true))
  val alive: Process[Task, Boolean] = Process.emitAll(k)

  val app = new Flask(options, new Instruments(1.minute))

  override def beforeAll() {
    app.run(Array())
    Http(flaskUrl << payload OK as.String)(concurrent.ExecutionContext.Implicits.global)
  }

  override def afterAll() {
    app.S.stop()
  }

  "mirrorDatapoints" should "be 5000" in {
    app.I.monitoring.get(app.mirrorDatapoints.keys.now).discrete.sleepUntil(ready.discrete.once).once.runLast.map(_.get).runAsync { d =>
      d.fold (
        t =>
        throw t,
        v =>
        v.toInt should be (5000)
      )
    }

    Ø.linkP(E)(alive)(socket =>
      proc.through(Ø.write(socket))).runFoldMap(identity).run
  }
}
<commit_msg>Explicitly return Unit.<commit_after>package funnel
package flask

import scala.concurrent.duration._
import scalaz.concurrent.{ Actor, Strategy, Task }
import scalaz.std.option._
import scalaz.syntax.applicative._
import org.scalatest.{FlatSpec,Matchers,BeforeAndAfterAll}
import scalaz.stream.Process
import scalaz.stream.async.signal
import com.amazonaws.auth.BasicAWSCredentials
import argonaut._, Argonaut._
import org.slf4j.LoggerFactory
import funnel.elastic._
import funnel.http.JSON._
import funnel.zeromq._
import sockets._

class FlaskSpec extends FlatSpec with Matchers with BeforeAndAfterAll {
  import java.io.File
  import knobs.{ ClassPathResource, Config, FileResource, Required }
  import dispatch._

  val config: Task[Config] = for {
    a <- knobs.loadImmutable(List(Required(
      FileResource(new File("/usr/share/oncue/etc/flask.cfg")) or
        ClassPathResource("oncue/flask.cfg"))))
    b <- knobs.aws.config
  } yield a ++ b

  val (options, cfg) = config.flatMap { cfg =>
    val name             = cfg.lookup[String]("flask.name").getOrElse("flask")
    val elasticURL       = cfg.lookup[String]("flask.elastic-search.url")
    val elasticIx        = cfg.lookup[String]("flask.elastic-search.index-name")
    val elasticTy        = cfg.lookup[String]("flask.elastic-search.type-name")
    val elasticDf        =
      cfg.lookup[String]("flask.elastic-search.partition-date-format").getOrElse("yyyy.MM.dd")
    val elasticTimeout   = cfg.lookup[Int]("flask.elastic-search.connection-timeout-in-ms").getOrElse(5000)
    val riemannHost      = cfg.lookup[String]("flask.riemann.host")
    val riemannPort      = cfg.lookup[Int]("flask.riemann.port")
    val ttl              = cfg.lookup[Int]("flask.riemann.ttl-in-minutes").map(_.minutes)
    val riemann          = (riemannHost |@| riemannPort |@| ttl)(RiemannCfg)
    val elastic          = (elasticURL |@| elasticIx |@| elasticTy)(
      ElasticCfg(_, _, _, elasticDf, elasticTimeout))
    val snsErrorTopic    = cfg.require[String]("flask.sns-error-topic")
    val awsAccessKey     = cfg.require[String]("aws.access-key")
    val awsSecretKey     = cfg.require[String]("aws.secret-key")
    val awsCredentials   = new BasicAWSCredentials(awsAccessKey, awsSecretKey)
    val awsProxyHost     = cfg.lookup[String]("aws.proxy-host")
    val awsProxyPort     = cfg.lookup[Int]("aws.proxy-port")
    val awsProxyProtocol = cfg.lookup[String]("aws.proxy-protocol")
    val awsRegion        = cfg.require[String]("aws.region")
    val port             = cfg.lookup[Int]("flask.network.port").getOrElse(5775)
    Task((Options(Option(name), elastic, riemann, snsErrorTopic, awsCredentials, awsProxyHost,
      awsProxyPort, awsProxyProtocol, awsRegion, port), cfg))
  }.run

  val logger = LoggerFactory.getLogger("flask")

  implicit val logPool: Strategy = Strategy.Executor(java.util.concurrent.Executors.newFixedThreadPool(1))

  val L = Actor.actor((s: String) => logger.info(s))

  implicit val log: String => Unit = s => L(s)

  val flaskUrl = url(s"http://localhost:${options.funnelPort}/mirror").setContentType("application/json", "UTF-8")

  val payload = s"""
  [
    {
      "bucket": "datapoints-1.0-us-east",
      "urls": [
        "${Settings.tcp}"
      ]
    }
  ]
  """

  val ready = signal[Boolean]
  ready.set(false)

  implicit val B = scalaz.std.anyVal.booleanInstance.conjunction
  implicit val s = scalaz.stream.DefaultScheduler

  val E = Endpoint.unsafeApply(publish &&& bind, Settings.tcp)

  implicit val batransport: Transportable[Array[Byte]] = Transportable { ba =>
    Transported(Schemes.unknown, Versions.v1, None, None, ba)
  }

  val seq: Seq[Array[Byte]] = for(i <- 0 until 5000) yield Datapoint(Key("now/life", Units.Count: Units[Double], "description", Map("url" -> "http://localhost")), 42.0).asJson.spaces2.getBytes
  val k: Seq[Boolean] = seq.map(_ => true) ++ Seq(false)

  val proc: Process[Task, Array[Byte]] = Process.emitAll(seq) fby Process.eval_(ready.set(true))
  val alive: Process[Task, Boolean] = Process.emitAll(k)

  val app = new Flask(options, new Instruments(1.minute))

  override def beforeAll() {
    app.run(Array())
    Http(flaskUrl << payload OK as.String)(concurrent.ExecutionContext.Implicits.global)
    ()
  }

  override def afterAll() {
    app.S.stop()
  }

  "mirrorDatapoints" should "be 5000" in {
    app.I.monitoring.get(app.mirrorDatapoints.keys.now).discrete.sleepUntil(ready.discrete.once).once.runLast.map(_.get).runAsync { d =>
      d.fold (
        t =>
        throw t,
        v =>
        v.toInt should be (5000)
      )
    }

    Ø.linkP(E)(alive)(socket =>
      proc.through(Ø.write(socket))).runFoldMap(identity).run
      ()
  }
}

{'fraction_non_alphanumeric': 0.14293327648688978, 'fraction_numerical': 0.00682157322532509, 'mean_word_length': 3.5726120857699804, 'pattern_counts': {'":': 4, '<': 13, '<?xml version=': 0, '>': 25, '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': '1992398', 'n_tokens_mistral': 3257, 'n_tokens_neox': 3092, 'n_words': 568}