Merge pull request #27 from amfern/http-sink

add http sink

Author: jippi

README.md

@@ -90,6 +90,8 @@ The `kafka` sink is configured using `$SINK_KAFKA_BROKERS` (`kafka1:9092,kafka2:
 
 The `mongo` sink is configured using `$SINK_MONGODB_CONNECTION` (`mongodb://localhost:27017/`), `$SINK_MONGODB_DATABASE` and `$SINK_MONGODB_COLLECTION` environment variables.
 
+The `http` sink is configured using `$SINK_HTTP_ADDRESS` (`localhost:8080/allocations`)` environment variable.
+
 The `stdout` sink does not have any configuration, it will simply output the JSON to stdout for debugging.
 
 ### `allocations`

sink/helper.go

@@ -9,7 +9,7 @@ import (
 func GetSink() (Sink, error) {
 	sinkType := os.Getenv("SINK_TYPE")
 	if sinkType == "" {
-		return nil, fmt.Errorf("Missing SINK_TYPE: amqp, kafka, kinesis, nsq, rabbitmq, redis or stdout")
+		return nil, fmt.Errorf("Missing SINK_TYPE: amqp, kafka, kinesis, nsq, rabbitmq, redis, mongodb, http or stdout")
 	}
 
 	switch sinkType {
@@ -27,9 +27,11 @@ func GetSink() (Sink, error) {
 		return NewRedis()
 	case "mongodb":
 		return NewMongodb()
+	case "http":
+		return NewHttp()
 	case "stdout":
 		return NewStdout()
 	default:
-		return nil, fmt.Errorf("Invalid SINK_TYPE: %s, Valid values: amqp, kafka, kinesis, nsq, rabbitmq, redis, mongodb or stdout", sinkType)
+		return nil, fmt.Errorf("Invalid SINK_TYPE: %s, Valid values: amqp, kafka, kinesis, nsq, rabbitmq, redis, mongodb, http or stdout", sinkType)
 	}
 }

sink/http.go

@@ -0,0 +1,103 @@
+package sink
+
+import (
+	"strconv"
+	"time"
+	"bytes"
+
+	"net/http"
+
+	"os"
+
+	"fmt"
+
+	log "github.com/sirupsen/logrus"
+)
+
+// HttpSink ...
+type HttpSink struct {
+	address     string
+	workerCount int
+	stopCh      chan interface{}
+	putCh       chan []byte
+}
+
+// NewHttp ...
+func NewHttp() (*HttpSink, error) {
+	address := os.Getenv("SINK_HTTP_ADDRESS")
+	if address == "" {
+		return nil, fmt.Errorf("[sink/http] Missing SINK_HTTP_ADDRESS (example: http://miau.com:8080/biau)")
+	}
+
+	workerCountStr := os.Getenv("SINK_WORKER_COUNT")
+	if workerCountStr == "" {
+		workerCountStr = "1"
+	}
+	workerCount, err := strconv.Atoi(workerCountStr)
+	if err != nil {
+		return nil, fmt.Errorf("Invalid SINK_WORKER_COUNT, must be an integer")
+	}
+
+	return &HttpSink{
+		address:     address,
+		workerCount: workerCount,
+		stopCh:      make(chan interface{}),
+		putCh:       make(chan []byte, 1000),
+	}, nil
+}
+
+// Start ...
+func (s *HttpSink) Start() error {
+	// Stop chan for all tasks to depend on
+	s.stopCh = make(chan interface{})
+
+	for i := 0; i < s.workerCount; i++ {
+		go s.send(i)
+	}
+
+	// wait forever for a stop signal to happen
+	for {
+		select {
+		case <-s.stopCh:
+			break
+		}
+		break
+	}
+
+	return nil
+}
+
+// Stop ...
+func (s *HttpSink) Stop() {
+	log.Infof("[sink/http] ensure writer queue is empty (%d messages left)", len(s.putCh))
+
+	for len(s.putCh) > 0 {
+		log.Info("[sink/http] Waiting for queue to drain - (%d messages left)", len(s.putCh))
+		time.Sleep(1 * time.Second)
+	}
+
+	close(s.stopCh)
+}
+
+// Put ..
+func (s *HttpSink) Put(data []byte) error {
+	s.putCh <- data
+
+	return nil
+}
+
+func (s *HttpSink) send(id int) {
+	log.Infof("[sink/http/%d] Starting writer", id)
+
+	for {
+		select {
+		case data := <-s.putCh:
+			_, err := http.Post(s.address, "text/json", bytes.NewReader(data))
+			if err != nil {
+				log.Errorf("[sink/http/%d] %s", id, err)
+			} else {
+				log.Debugf("[sink/http/%d] publish ok", id)
+			}
+		}
+	}
+}