NSQ input handler for heurelog

nsq.go

package input

import (
	"sync"

	"fmt"

	"github.com/PumpkinSeed/heurelog/config"
	"github.com/PumpkinSeed/heurelog/handlers/output"
	"github.com/Sirupsen/logrus"
	nsq "github.com/nsqio/go-nsq"
)

type NSQ struct {
	C      config.Interface
	Output output.Interface
}

func NewNSQ(cnf config.Interface, out output.Interface) (*NSQ, error) {
	return &NSQ{
		C:      cnf,
		Output: out,
	}, nil
}

func (n *NSQ) Handle() {
	conf := n.C.GetInput().NSQ
	log := logrus.New()
	wg := &sync.WaitGroup{}
	wg.Add(1)

	config := nsq.NewConfig()
	q, _ := nsq.NewConsumer(conf.Topic, conf.Channel, config)
	q.AddHandler(nsq.HandlerFunc(func(message *nsq.Message) error {
		log.Printf("Got a message: %s", string(message.Body))
		n.Output.Write(message.Body)
		return nil
	}))
	err := q.ConnectToNSQD(fmt.Sprintf("%s:%s", conf.Host, conf.Port))
	if err != nil {
		log.Panic("Could not connect")
	}
	wg.Wait()

	log.Info("NSQ server started to consume")
}