package main

import (
  "flag"
  "fmt"
  "github.com/golang/protobuf/proto"
  "github.com/bradhe/epee"
)

const (
  // This is the topic that the stream processor will listen to.
  TopicName = "my-topic"

  // This is the partition that the processor will listen on.
  Partition = 1

  // NOTE: This must be unique to the topic and partition the stream processor is 
  // going to consume.
  DefaultClientID = "my-client-1"
)

var (
  stream *epee.Stream
  
  // Parameterize where Zookeeper lives.
  ZookeeperHost = flag.String("zookeeper-host", "localhost:2181", "zookeeper host")
)

// This type encapsulates the stream processor and will implement the
// epee.StreamProcessor interface.
type MyStreamProcessor struct {
  Total int64
}

// The process method is called once for each message in the queue. If the message
// is successfully processed the related offset will be marked as "processed" so
// that when clients resume later this message doesn't get re-processed.
func (sp *MyStreamProcessor) Process(offset int64, message proto.Message) error {
  counter, ok := message.(*MyCounter)

  if !ok {
    return fmt.Errorf("failed to convert message to application-native type")	
  }

  sp.Total += counter.GetCount()
  return nil
}

// The flush method will be periodically called (once every 10 seconds by
// default). This method is used to flush the processor's state so the jobs can
// be resumed if something goes wrong.
func (sp *MyStreamProcessor) Flush() error {
  // TODO: Flush the total to something here.
  return nil
}

func init() {
  flag.Parse()

  zk, err := epee.NewZookeeperClient(*ZookeeperHost)

  if err != nil {
    panic(err)
  }
  
  // Assuming your Kafka brokers are registered in Zookeeper...
  stream, err = epee.NewStream(DefaultClientID, zk)

  if err != nil {
    panic(err)
  }

  // This tells the stream how to deserialize the message in Kafka.
  stream.Register(TopicName, &MyCounter{})
}

func main() {
  stream.Stream(TopicName, Partition, &MyStreamProcessor{})

  // The stream processor is now running in a goroutine in the background. The
  // main thread can continue doing whatever, or we can just sit here and wait.
  stream.Wait()
}