Initial import.

pubsub/publisher.go

@@ -0,0 +1,58 @@
+package pubsub
+
+import (
+	"time"
+
+	telemetrypb "git.wntrmute.dev/kyle/sensenet/proto"
+	"google.golang.org/protobuf/proto"
+	goczmq "gopkg.in/zeromq/goczmq.v4"
+)
+
+const singleFrame = 0
+
+type Publisher struct {
+	addr string
+	sock *goczmq.Sock
+}
+
+func NewPublisher(addr string) (*Publisher, error) {
+	pub := &Publisher{
+		addr: addr,
+		sock: goczmq.NewSock(goczmq.Pub),
+	}
+
+	pub.Conflate(1)
+	err := pub.connect()
+	if err != nil {
+		return nil, err
+	}
+
+	return pub, nil
+}
+
+func (pub *Publisher) connect() error {
+	return pub.sock.Connect(pub.addr)
+}
+
+func (pub *Publisher) Conflate(n int) {
+	pub.sock.SetConflate(n)
+}
+
+func (pub *Publisher) Transmit(topic string, data []byte) error {
+	packet, err := genPacket(topic, data)
+	if err != nil {
+		return err
+	}
+
+	return pub.sock.SendFrame(packet, singleFrame)
+}
+
+func genPacket(topic string, payload []byte) ([]byte, error) {
+	packet := &telemetrypb.Packet{
+		Topic:     topic,
+		Timestamp: uint64(time.Now().Unix()),
+		Payload:   payload,
+	}
+
+	return proto.Marshal(packet)
+}

pubsub/subscriber.go

@@ -0,0 +1,105 @@
+package pubsub
+
+import (
+	"fmt"
+	"sort"
+
+	telemetrypb "git.wntrmute.dev/kyle/sensenet/proto"
+	"git.wntrmute.dev/kyle/sensenet/topic"
+	"google.golang.org/protobuf/proto"
+	"gopkg.in/zeromq/goczmq.v4"
+)
+
+type Subscriber struct {
+	addr      string
+	publisher string
+	sock      *goczmq.Sock
+	topics    map[string]bool
+}
+
+func NewSubscriber(addr, publisher string, topics ...string) (*Subscriber, error) {
+	sub := &Subscriber{
+		addr: addr,
+		sock: goczmq.NewSock(goczmq.Pub),
+	}
+
+	err := sub.connect()
+	if err != nil {
+		return nil, err
+	}
+
+	sub.Conflate(1)
+	for _, topic := range topics {
+		sub.Subscribe(topic)
+	}
+
+	return sub, nil
+}
+
+func (sub *Subscriber) Conflate(n int) {
+	sub.sock.SetConflate(n)
+}
+
+func (sub *Subscriber) connect() error {
+	return sub.sock.Connect(sub.addr)
+}
+
+func (sub *Subscriber) Subscribe(topic string) {
+	if sub.topics[topic] {
+		return
+	}
+
+	sub.topics[topic] = true
+	topic = fmt.Sprintf("\x0a%d%s", len(topic), topic)
+	sub.sock.SetSubscribe(topic)
+}
+
+func (sub *Subscriber) Topics() []string {
+	var topics = make([]string, 0, len(sub.topics))
+	for k := range sub.topics {
+		topics = append(topics, k)
+	}
+
+	sort.Strings(topics)
+	return topics
+}
+
+func (sub *Subscriber) receive() ([]byte, error) {
+	var (
+		data   []byte
+		err    error
+		packet []byte
+		more   = 1
+	)
+	for more > 0 {
+		packet, more, err = sub.sock.RecvFrame()
+		if err != nil {
+			return nil, err
+		}
+
+		data = append(data, packet...)
+	}
+
+	return data, nil
+}
+
+func (sub *Subscriber) Receive() (*topic.Packet, error) {
+	data, err := sub.receive()
+	if err != nil {
+		return nil, err
+	}
+
+	pbPacket := &telemetrypb.Packet{}
+	err = proto.Unmarshal(data, pbPacket)
+	if err != nil {
+		return nil, err
+	}
+
+	packet := &topic.Packet{
+		Publisher: sub.publisher,
+		Received:  int64(pbPacket.Timestamp),
+		Payload:   pbPacket.Payload,
+	}
+
+	return packet, nil
+}