ircd/irc/cluster.go

68 lines
1.3 KiB
Go

// vim:ts=4:sts=4:sw=4:noet:tw=72
package irc
import (
"strings"
"git.dnix.de/an/xlog"
nats "github.com/nats-io/nats.go"
)
type ClusterConnector struct {
conn *nats.Conn
subs map[string]*nats.Subscription
}
func NewClusterConnector(urls string, ssl bool) *ClusterConnector {
opts := nats.DefaultOptions
opts.Servers = strings.Split(urls, ",")
for i, s := range opts.Servers {
opts.Servers[i] = strings.Trim(s, " ")
}
opts.Secure = ssl
conn, err := opts.Connect()
if err != nil {
xlog.Error(err.Error())
}
subs := make(map[string]*nats.Subscription)
return &ClusterConnector{conn: conn, subs: subs}
}
func (cc *ClusterConnector) Subscribe(subj string, ch chan *Message) {
if cc.conn == nil {
return
}
if _, exists := cc.subs[subj]; exists {
return
}
sub, err := cc.conn.Subscribe(subj, func(n *nats.Msg) {
m := Parse(string(n.Data))
ch <- m
})
if err != nil {
xlog.Error(err.Error())
return
}
cc.subs[subj] = sub
}
func (cc *ClusterConnector) Unsubscribe(subj string) {
if cc.conn == nil {
return
}
if sub, exists := cc.subs[subj]; exists {
sub.Unsubscribe()
delete(cc.subs, subj)
}
}
func (cc *ClusterConnector) Publish(msg *Message) {
if cc.conn == nil {
return
}
subj := strings.ToLower(msg.Args[0])
cc.conn.Publish(subj, []byte(msg.String()))
}