first commit.

way too late.

nanomsgsubscriber/nanomsgsubscriber.go

@@ -0,0 +1,59 @@
+package nanomsgsubscriber
+
+import (
+	"encoding/json"
+	"errors"
+	"log"
+
+	"github.com/didiercrunch/doorman/shared"
+	"github.com/gdamore/mangos"
+	"github.com/gdamore/mangos/protocol/sub"
+	"github.com/gdamore/mangos/transport/ipc"
+	"github.com/gdamore/mangos/transport/tcp"
+	"gopkg.in/mgo.v2/bson"
+)
+
+type NanoMsgSubscriber struct {
+	Url string
+}
+
+func (s *NanoMsgSubscriber) callUpdateHandlerFunction(f shared.UpdateHandlerFunc, data []byte) error {
+	wu := &shared.DoormanUpdater{}
+	if err := json.Unmarshal(data, wu); err != nil {
+		return err
+	} else {
+		return f(wu)
+	}
+}
+
+func (s *NanoMsgSubscriber) Subscribe(abtestId bson.ObjectId, update shared.UpdateHandlerFunc) error {
+	var sock mangos.Socket
+	var err error
+	var msg []byte
+
+	if sock, err = sub.NewSocket(); err != nil {
+		return errors.New("can't get new sub socket: " + err.Error())
+	}
+	sock.AddTransport(ipc.NewTransport())
+	sock.AddTransport(tcp.NewTransport())
+	if err = sock.Dial(s.Url); err != nil {
+		return errors.New("can't dial on sub socket: " + err.Error())
+	}
+	// Empty byte array effectively subscribes to everything
+	if err = sock.SetOption(mangos.OptionSubscribe, []byte("")); err != nil {
+		return errors.New("cannot subscribe: " + err.Error())
+	}
+	go func(s *NanoMsgSubscriber, sock mangos.Socket) {
+		for {
+			if msg, err = sock.Recv(); err != nil {
+				err := errors.New("Cannot recv: " + err.Error())
+				log.Println(err)
+			} else if err := s.callUpdateHandlerFunction(update, msg); err != nil {
+				log.Println("cannot update abtest with received data: ", err)
+			}
+		}
+	}(s, sock)
+
+	return nil
+
+}