<commit_before>package intelmedia.ws
package monitoring 

import scopt.{OptionParser,Read}
import java.net.URL
import scala.concurrent.duration._

case class Options(urls: Seq[URL], duration: Duration = 5 minutes)

trait FunnelCommandLine {
  implicit val scoptReadUrl: Read[URL] = Read.reads { new URL(_) }
  implicit val scoptReadDuration: Read[Duration] = Read.reads { Duration(_) }

  // protected val build = new BuildData
  protected val parser = new OptionParser[Options]("funnel"){
    head("funnel", "1.0")
    
    opt[Duration]('d',"duration").action { (time, opts) =>
      opts.copy(duration = time)
    }.text("Duration is a required property")
    
    arg[URL]("<url>[,url]...") unbounded() optional() action { (url, opts) =>
      opts.copy(urls = opts.urls :+ url) } text("optional unbounded args")
  }

  def run(args: Array[String])(f: Options => Unit): Unit = 
    parser.parse(args, Options(Seq.empty)).foreach(f)
}

import Monitoring.default.mirrorAndAggregate
import scalaz.stream.Process

object Funnel extends FunnelCommandLine {
  def main(args: Array[String]): Unit = 
    run(args){ options =>  
      val health = Key[Boolean]("now/health", Units.Healthy)

      mirrorAndAggregate(Events.takeEvery(options.duration, 5))(Process.emitAll(options.urls), health) {
        case "accounts" => Policies.quorum(2)
        case "decoding" => Policies.majority
        case _          => sys.error("unknown group type")
      }.run
    }
} <commit_msg>first pass at the command line cli; seems to have bugs?<commit_after>package intelmedia.ws
package monitoring 

import scopt.{OptionParser,Read}
import java.net.URL
import scala.concurrent.duration._

case class Options(
  urls: Seq[(URL,String)], 
  aggregationInterval: Duration = 5 minutes,
  retryLimit: Int = 3,
  healthKey: String = "now/health")

trait FunnelCommandLine {
  // takes strings of the form:
  // accounts@http://foobar.com
  private def splitBucketURLPairs(str: String): (URL,String) = 
    str.split('@') match {
      case Array(bucket,urlstring) => (new URL(urlstring),bucket)
      case _ => sys.error(s"Unable to parse the supplied bucket@url pair: $str")
    }

  implicit val scoptReadUrl: Read[(URL,String)] = Read.reads { splitBucketURLPairs(_) }
  implicit val scoptReadDuration: Read[Duration] = Read.reads { Duration(_) }

  // protected val build = new BuildData
  protected val parser = new OptionParser[Options]("funnel"){
    head("funnel", "1.0")
    
    opt[Duration]('i',"interval").action { (duration, opts) =>
      opts.copy(aggregationInterval = duration)
    }

    opt[Int]('r',"retries").action { (limit, opts) => 
      opts.copy(retryLimit = limit)
    }

    arg[(URL,String)]("<url>[,url]...") unbounded() optional() action { (url, opts) =>
      opts.copy(urls = opts.urls :+ url) } text("optional unbounded args")
  }

  def run(args: Array[String])(f: Options => Unit): Unit = 
    parser.parse(args, Options(Seq.empty)).foreach(f)
}

import Monitoring.default.mirrorAndAggregate
import scalaz.stream.Process

object Funnel extends FunnelCommandLine {
  def main(args: Array[String]): Unit = 
    run(args){ options =>  
      println(">>>> "+ options)
      val health = Key[Boolean](options.healthKey, Units.Healthy)
      val events = Events.takeEvery(options.aggregationInterval, options.retryLimit)

      mirrorAndAggregate(events)(Process.emitAll(options.urls), health) {
        case "accounts" => Policies.quorum(2)
        case "test"     => Policies.majority
        case _          => sys.error("unknown group type")
      }.run

      println
      println("Press [Enter] to stop the funnel...")
      println

      readLine()
    }
} 
{'fraction_non_alphanumeric': 0.14192495921696574, 'fraction_numerical': 0.0027188689505165853, 'mean_word_length': 3.6452020202020203, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 26, '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': '7369124', 'n_tokens_mistral': 1173, 'n_tokens_neox': 1136, 'n_words': 270}