ActionSeedFileCLI.scala

package com.yieldmo.tools

import com.google.common.cache._
import com.yieldmo.common.DBUtil
import com.yieldmo.storm.StateSaver
import com.yieldmo.common.protobuf.Events._
import com.github.theon.uri.Uri._
import kafka.producer.ProducerConfig
import scala.collection.JavaConversions._
import scala.io.Source
import scopt.mutable.OptionParser
import java.util.concurrent.TimeUnit



object Publisher {

  // max 10K placementId's
  val cacheBlock: LoadingCache[java.lang.Long, java.lang.Long] = CacheBuilder
    .newBuilder()
    .maximumSize(10000)
    .expireAfterAccess(1, TimeUnit.HOURS)
    .build(new CacheLoader[java.lang.Long, java.lang.Long] {
      val sql = "SELECT publisher_id FROM placement WHERE placement_id = ?";
      override def load(placementId: java.lang.Long): java.lang.Long = {
        val objs  = DBUtil.queryDB(sql,placementId)
        try {
          objs(0)(0).asInstanceOf[java.lang.Long]
        }catch {
          case e: Exception =>
            println("error trying to get the publisher for placementId:" + placementId)
            new java.lang.Long(0);
        }

      }
    })

  def getPub(placementId: Long) = cacheBlock.get(new java.lang.Long(placementId))

}

object ActionSeedFileCLI extends App{
  var file: String = null
  val parser = new OptionParser("ActionSeedFileCLI", true){
    arg("<file>", "File to read", { (arg : String) => file = arg })
  }

  implicit def toLong(s:String) = s.toLong

  if(parser.parse(args)) {
    val save = new StateSaver()
    Source.fromFile(file).getLines().foreach { line =>
      try {
        val uri = parseUri("http://localhost/?"+line)
        val params = uri.query.params
        val ab = ActionSeed.newBuilder()

        (params.get("type"),
          params.get("stime"),
          params.get("etime"),
          params.get("pvid"),
          params.get("plid"),
          params.get("pvt")) match {

          case (Some((t:String) :: _),
            Some((startTime:String) :: _),
            Some((endTime:String) :: _),
            Some((pageViewId:String) :: _),
            Some((placementId:String) :: _ ),
            Some((pageViewTime:String) :: _)) =>
            ab.setType({
              t match {
                case "v" => ActionSeed.Type.VISIBLE
                case "d" => ActionSeed.Type.DURATION
              }
            })
            ab.setPageViewId(pageViewId)
            ab.setPageViewTime(pageViewTime)
            ab.setPlacementId(placementId)
            ab.setStartTime(startTime)
            ab.setEndTime(endTime)
            ab.setPublisherId(Publisher.getPub(placementId))
            save.saveBulk(List(ab.build))
          case _ =>
            println("Could not parse line:" + line)
        }
      }catch {
        case e: Exception => println("Could not parse line:[" + line + "] with error: " + e)
      }
    }
  }

  println("Done!")
  System.exit(0);
}