parent
0db2ab6b97
commit
354a101d26
6
go.mod
6
go.mod
|
@ -4,14 +4,15 @@ go 1.17
|
||||||
|
|
||||||
require (
|
require (
|
||||||
github.com/Masterminds/squirrel v1.5.2
|
github.com/Masterminds/squirrel v1.5.2
|
||||||
|
github.com/go-zeromq/zmq4 v0.15.0
|
||||||
github.com/jackc/pgconn v1.11.0
|
github.com/jackc/pgconn v1.11.0
|
||||||
github.com/jackc/pgx/v4 v4.11.0
|
github.com/jackc/pgx/v4 v4.11.0
|
||||||
google.golang.org/protobuf v1.27.1
|
google.golang.org/protobuf v1.27.1
|
||||||
gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c
|
gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c
|
||||||
gopkg.in/zeromq/goczmq.v4 v4.1.0
|
|
||||||
)
|
)
|
||||||
|
|
||||||
require (
|
require (
|
||||||
|
github.com/go-zeromq/goczmq/v4 v4.2.2 // indirect
|
||||||
github.com/jackc/chunkreader/v2 v2.0.1 // indirect
|
github.com/jackc/chunkreader/v2 v2.0.1 // indirect
|
||||||
github.com/jackc/pgio v1.0.0 // indirect
|
github.com/jackc/pgio v1.0.0 // indirect
|
||||||
github.com/jackc/pgpassfile v1.0.0 // indirect
|
github.com/jackc/pgpassfile v1.0.0 // indirect
|
||||||
|
@ -22,5 +23,6 @@ require (
|
||||||
github.com/lann/builder v0.0.0-20180802200727-47ae307949d0 // indirect
|
github.com/lann/builder v0.0.0-20180802200727-47ae307949d0 // indirect
|
||||||
github.com/lann/ps v0.0.0-20150810152359-62de8c46ede0 // indirect
|
github.com/lann/ps v0.0.0-20150810152359-62de8c46ede0 // indirect
|
||||||
golang.org/x/crypto v0.0.0-20210711020723-a769d52b0f97 // indirect
|
golang.org/x/crypto v0.0.0-20210711020723-a769d52b0f97 // indirect
|
||||||
golang.org/x/text v0.3.6 // indirect
|
golang.org/x/sync v0.0.0-20220601150217-0de741cfad7f // indirect
|
||||||
|
golang.org/x/text v0.3.7 // indirect
|
||||||
)
|
)
|
||||||
|
|
11
go.sum
11
go.sum
|
@ -68,6 +68,10 @@ github.com/go-logfmt/logfmt v0.4.0/go.mod h1:3RMwSq7FuexP4Kalkev3ejPJsZTpXXBr9+V
|
||||||
github.com/go-logfmt/logfmt v0.5.0/go.mod h1:wCYkCAKZfumFQihp8CzCvQ3paCTfi41vtzG1KdI/P7A=
|
github.com/go-logfmt/logfmt v0.5.0/go.mod h1:wCYkCAKZfumFQihp8CzCvQ3paCTfi41vtzG1KdI/P7A=
|
||||||
github.com/go-sql-driver/mysql v1.4.0/go.mod h1:zAC/RDZ24gD3HViQzih4MyKcchzm+sOG5ZlKdlhCg5w=
|
github.com/go-sql-driver/mysql v1.4.0/go.mod h1:zAC/RDZ24gD3HViQzih4MyKcchzm+sOG5ZlKdlhCg5w=
|
||||||
github.com/go-stack/stack v1.8.0/go.mod h1:v0f6uXyyMGvRgIKkXu+yp6POWl0qKG85gN/melR3HDY=
|
github.com/go-stack/stack v1.8.0/go.mod h1:v0f6uXyyMGvRgIKkXu+yp6POWl0qKG85gN/melR3HDY=
|
||||||
|
github.com/go-zeromq/goczmq/v4 v4.2.2 h1:HAJN+i+3NW55ijMJJhk7oWxHKXgAuSBkoFfvr8bYj4U=
|
||||||
|
github.com/go-zeromq/goczmq/v4 v4.2.2/go.mod h1:Sm/lxrfxP/Oxqs0tnHD6WAhwkWrx+S+1MRrKzcxoaYE=
|
||||||
|
github.com/go-zeromq/zmq4 v0.15.0 h1:SLqukpmLTx0JsLaOaCCjwy5eBdfJ+ouJX/677HoFbJM=
|
||||||
|
github.com/go-zeromq/zmq4 v0.15.0/go.mod h1:sD47DcXifeUFsVTB2ps8ijqTpEuTAlYgfuLoiWEXdCE=
|
||||||
github.com/gofrs/uuid v3.2.0+incompatible h1:y12jRkkFxsd7GpqdSZ+/KCs/fJbqpEXSGd4+jfEaewE=
|
github.com/gofrs/uuid v3.2.0+incompatible h1:y12jRkkFxsd7GpqdSZ+/KCs/fJbqpEXSGd4+jfEaewE=
|
||||||
github.com/gofrs/uuid v3.2.0+incompatible/go.mod h1:b2aQJv3Z4Fp6yNu3cdSllBxTCLRxnplIgP/c0N/04lM=
|
github.com/gofrs/uuid v3.2.0+incompatible/go.mod h1:b2aQJv3Z4Fp6yNu3cdSllBxTCLRxnplIgP/c0N/04lM=
|
||||||
github.com/gogo/googleapis v1.1.0/go.mod h1:gf4bu3Q80BeJ6H1S1vYPm8/ELATdvryBaNFGgqEef3s=
|
github.com/gogo/googleapis v1.1.0/go.mod h1:gf4bu3Q80BeJ6H1S1vYPm8/ELATdvryBaNFGgqEef3s=
|
||||||
|
@ -397,6 +401,8 @@ golang.org/x/sync v0.0.0-20181221193216-37e7f081c4d4/go.mod h1:RxMgew5VJxzue5/jJ
|
||||||
golang.org/x/sync v0.0.0-20190227155943-e225da77a7e6/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
|
golang.org/x/sync v0.0.0-20190227155943-e225da77a7e6/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
|
||||||
golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
|
golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
|
||||||
golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
|
golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
|
||||||
|
golang.org/x/sync v0.0.0-20220601150217-0de741cfad7f h1:Ax0t5p6N38Ga0dThY21weqDEyz2oklo4IvDkpigvkD8=
|
||||||
|
golang.org/x/sync v0.0.0-20220601150217-0de741cfad7f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
|
||||||
golang.org/x/sys v0.0.0-20180823144017-11551d06cbcc/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
|
golang.org/x/sys v0.0.0-20180823144017-11551d06cbcc/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
|
||||||
golang.org/x/sys v0.0.0-20180830151530-49385e6e1522/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
|
golang.org/x/sys v0.0.0-20180830151530-49385e6e1522/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
|
||||||
golang.org/x/sys v0.0.0-20180905080454-ebe1bf3edb33/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
|
golang.org/x/sys v0.0.0-20180905080454-ebe1bf3edb33/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
|
||||||
|
@ -426,8 +432,9 @@ golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
|
||||||
golang.org/x/text v0.3.2/go.mod h1:bEr9sfX3Q8Zfm5fL9x+3itogRgK3+ptLWKqgva+5dAk=
|
golang.org/x/text v0.3.2/go.mod h1:bEr9sfX3Q8Zfm5fL9x+3itogRgK3+ptLWKqgva+5dAk=
|
||||||
golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
|
golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
|
||||||
golang.org/x/text v0.3.4/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
|
golang.org/x/text v0.3.4/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
|
||||||
golang.org/x/text v0.3.6 h1:aRYxNxv6iGQlyVaZmk6ZgYEDa+Jg18DxebPSrd6bg1M=
|
|
||||||
golang.org/x/text v0.3.6/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
|
golang.org/x/text v0.3.6/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
|
||||||
|
golang.org/x/text v0.3.7 h1:olpwvP2KacW1ZWvsR7uQhoyTYvKAupfQrRGBFM352Gk=
|
||||||
|
golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ=
|
||||||
golang.org/x/time v0.0.0-20180412165947-fbb02b2291d2/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
|
golang.org/x/time v0.0.0-20180412165947-fbb02b2291d2/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
|
||||||
golang.org/x/time v0.0.0-20191024005414-555d28b269f0/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
|
golang.org/x/time v0.0.0-20191024005414-555d28b269f0/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
|
||||||
golang.org/x/tools v0.0.0-20180221164845-07fd8470d635/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
|
golang.org/x/tools v0.0.0-20180221164845-07fd8470d635/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
|
||||||
|
@ -490,8 +497,6 @@ gopkg.in/yaml.v2 v2.2.1/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
|
||||||
gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
|
gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
|
||||||
gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c h1:dUUwHk2QECo/6vqA44rthZ8ie2QXMNeKRTHCNY2nXvo=
|
gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c h1:dUUwHk2QECo/6vqA44rthZ8ie2QXMNeKRTHCNY2nXvo=
|
||||||
gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
|
gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
|
||||||
gopkg.in/zeromq/goczmq.v4 v4.1.0 h1:CE+FE81mGVs2aSlnbfLuS1oAwdcVywyMM2AC1g33imI=
|
|
||||||
gopkg.in/zeromq/goczmq.v4 v4.1.0/go.mod h1:h4IlfePEYMpFdywGr5gAwKhBBj+hiBl/nF4VoSE4k+0=
|
|
||||||
honnef.co/go/tools v0.0.0-20180728063816-88497007e858/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=
|
honnef.co/go/tools v0.0.0-20180728063816-88497007e858/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=
|
||||||
honnef.co/go/tools v0.0.0-20190102054323-c2f93a96b099/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=
|
honnef.co/go/tools v0.0.0-20190102054323-c2f93a96b099/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=
|
||||||
honnef.co/go/tools v0.0.0-20190523083050-ea95bdfd59fc/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=
|
honnef.co/go/tools v0.0.0-20190523083050-ea95bdfd59fc/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=
|
||||||
|
|
|
@ -1,24 +1,23 @@
|
||||||
package pubsub
|
package pubsub
|
||||||
|
|
||||||
import (
|
import (
|
||||||
|
"context"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
telemetrypb "git.wntrmute.dev/kyle/sensenet/proto"
|
telemetrypb "git.wntrmute.dev/kyle/sensenet/proto"
|
||||||
|
"github.com/go-zeromq/zmq4"
|
||||||
"google.golang.org/protobuf/proto"
|
"google.golang.org/protobuf/proto"
|
||||||
goczmq "gopkg.in/zeromq/goczmq.v4"
|
|
||||||
)
|
)
|
||||||
|
|
||||||
const singleFrame = 0
|
|
||||||
|
|
||||||
type Publisher struct {
|
type Publisher struct {
|
||||||
addr string
|
addr string
|
||||||
sock *goczmq.Sock
|
sock zmq4.Socket
|
||||||
}
|
}
|
||||||
|
|
||||||
func NewPublisher(addr string) (*Publisher, error) {
|
func NewPublisher(addr string) (*Publisher, error) {
|
||||||
pub := &Publisher{
|
pub := &Publisher{
|
||||||
addr: addr,
|
addr: addr,
|
||||||
sock: goczmq.NewSock(goczmq.Pub),
|
sock: zmq4.NewPub(context.Background()),
|
||||||
}
|
}
|
||||||
|
|
||||||
pub.Conflate(1)
|
pub.Conflate(1)
|
||||||
|
@ -31,11 +30,11 @@ func NewPublisher(addr string) (*Publisher, error) {
|
||||||
}
|
}
|
||||||
|
|
||||||
func (pub *Publisher) connect() error {
|
func (pub *Publisher) connect() error {
|
||||||
return pub.sock.Connect(pub.addr)
|
return pub.sock.Dial(pub.addr)
|
||||||
}
|
}
|
||||||
|
|
||||||
func (pub *Publisher) Conflate(n int) {
|
func (pub *Publisher) Conflate(n int) {
|
||||||
pub.sock.SetConflate(n)
|
pub.sock.SetOption("CONFLATE", n)
|
||||||
}
|
}
|
||||||
|
|
||||||
func (pub *Publisher) Transmit(topic string, data []byte) error {
|
func (pub *Publisher) Transmit(topic string, data []byte) error {
|
||||||
|
@ -44,7 +43,11 @@ func (pub *Publisher) Transmit(topic string, data []byte) error {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
return pub.sock.SendFrame(packet, singleFrame)
|
msg := zmq4.Msg{
|
||||||
|
Frames: [][]byte{packet},
|
||||||
|
}
|
||||||
|
|
||||||
|
return pub.sock.Send(msg)
|
||||||
}
|
}
|
||||||
|
|
||||||
func genPacket(topic string, payload []byte) ([]byte, error) {
|
func genPacket(topic string, payload []byte) ([]byte, error) {
|
||||||
|
|
|
@ -1,14 +1,15 @@
|
||||||
package pubsub
|
package pubsub
|
||||||
|
|
||||||
import (
|
import (
|
||||||
|
"context"
|
||||||
"fmt"
|
"fmt"
|
||||||
"log"
|
"log"
|
||||||
"sort"
|
"sort"
|
||||||
|
|
||||||
telemetrypb "git.wntrmute.dev/kyle/sensenet/proto"
|
telemetrypb "git.wntrmute.dev/kyle/sensenet/proto"
|
||||||
"git.wntrmute.dev/kyle/sensenet/topic"
|
"git.wntrmute.dev/kyle/sensenet/topic"
|
||||||
|
"github.com/go-zeromq/zmq4"
|
||||||
"google.golang.org/protobuf/proto"
|
"google.golang.org/protobuf/proto"
|
||||||
"gopkg.in/zeromq/goczmq.v4"
|
|
||||||
)
|
)
|
||||||
|
|
||||||
func readTopic(b []byte) string {
|
func readTopic(b []byte) string {
|
||||||
|
@ -23,7 +24,7 @@ func protoTopic(topic string) string {
|
||||||
type Subscriber struct {
|
type Subscriber struct {
|
||||||
addr string
|
addr string
|
||||||
publisher string
|
publisher string
|
||||||
sock *goczmq.Sock
|
sock zmq4.Socket
|
||||||
topics map[string]bool
|
topics map[string]bool
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -31,7 +32,7 @@ func NewSubscriber(addr, publisher string, topics ...string) (*Subscriber, error
|
||||||
sub := &Subscriber{
|
sub := &Subscriber{
|
||||||
addr: addr,
|
addr: addr,
|
||||||
publisher: publisher,
|
publisher: publisher,
|
||||||
sock: goczmq.NewSock(goczmq.Sub),
|
sock: zmq4.NewSub(context.Background()),
|
||||||
topics: map[string]bool{},
|
topics: map[string]bool{},
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -49,12 +50,12 @@ func NewSubscriber(addr, publisher string, topics ...string) (*Subscriber, error
|
||||||
}
|
}
|
||||||
|
|
||||||
func (sub *Subscriber) Conflate(n int) {
|
func (sub *Subscriber) Conflate(n int) {
|
||||||
sub.sock.SetConflate(n)
|
sub.sock.SetOption("CONFLATE", n)
|
||||||
}
|
}
|
||||||
|
|
||||||
func (sub *Subscriber) connect() error {
|
func (sub *Subscriber) connect() error {
|
||||||
log.Printf("subscriber dialing %s", sub.addr)
|
log.Printf("subscriber dialing %s", sub.addr)
|
||||||
return sub.sock.Connect(sub.addr)
|
return sub.sock.Listen(sub.addr)
|
||||||
}
|
}
|
||||||
|
|
||||||
func (sub *Subscriber) Subscribe(topic string) {
|
func (sub *Subscriber) Subscribe(topic string) {
|
||||||
|
@ -63,7 +64,7 @@ func (sub *Subscriber) Subscribe(topic string) {
|
||||||
}
|
}
|
||||||
|
|
||||||
sub.topics[topic] = true
|
sub.topics[topic] = true
|
||||||
sub.sock.SetSubscribe(protoTopic(topic))
|
sub.sock.SetOption(zmq4.OptionSubscribe, protoTopic(topic))
|
||||||
}
|
}
|
||||||
|
|
||||||
func (sub *Subscriber) Topics() []string {
|
func (sub *Subscriber) Topics() []string {
|
||||||
|
@ -77,19 +78,15 @@ func (sub *Subscriber) Topics() []string {
|
||||||
}
|
}
|
||||||
|
|
||||||
func (sub *Subscriber) receive() ([]byte, error) {
|
func (sub *Subscriber) receive() ([]byte, error) {
|
||||||
var (
|
var data []byte
|
||||||
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...)
|
msg, err := sub.sock.Recv()
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
|
||||||
|
for _, frame := range msg.Frames {
|
||||||
|
data = append(data, frame...)
|
||||||
}
|
}
|
||||||
|
|
||||||
return data, nil
|
return data, nil
|
||||||
|
|
Loading…
Reference in New Issue