ircd/cluster.go

// vim:ts=4:sts=4:sw=4:noet:tw=72

package ircd

import (
	"strings"

	"code.dnix.de/an/irc"
	"code.dnix.de/an/xlog"

	"github.com/nats-io/nats"
)

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 *irc.Message) {
	if cc.conn == nil {
		return
	}
	if _, exists := cc.subs[subj]; exists {
		return
	}
	sub, err := cc.conn.Subscribe(subj, func(n *nats.Msg) {
		m := irc.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 *irc.Message) {
	if cc.conn == nil {
		return
	}
	subj := strings.ToLower(msg.Args[0])
	cc.conn.Publish(subj, []byte(msg.String()))
}