ProtoScheme.scala

package com.yieldmo.storm.scheme

import backtype.storm.spout.Scheme
import backtype.storm.tuple.Fields
import com.google.protobuf._
import collection.JavaConversions._

class ProtoScheme[T <: Message: ClassManifest](emittedField: String) extends Scheme {
  val serialVersionUID = 1L;
  @transient lazy val builder =
    classManifest[T].erasure.getMethods.find(_.getName == "newBuilder").map(m => m.invoke(m.getDefaultValue).asInstanceOf[Message.Builder]).get
  override def deserialize(bytes: Array[Byte]) = asJavaList(List[Object](builder.mergeFrom(bytes).build()))
  override def getOutputFields: Fields = new Fields(emittedField)
}