micro-broker-nats/nats.go

222 lines
3.9 KiB
Go
Raw Normal View History

2017-01-01 23:29:12 +03:00
// Package nats provides a NATS broker
2015-11-25 03:16:12 +03:00
package nats
import (
2018-03-03 15:28:44 +03:00
"context"
2015-11-25 03:16:12 +03:00
"strings"
2018-11-05 09:39:28 +03:00
"sync"
2015-11-25 03:16:12 +03:00
"github.com/micro/go-micro/broker"
2019-01-10 12:42:14 +03:00
"github.com/micro/go-micro/codec/json"
2015-11-25 03:16:12 +03:00
"github.com/micro/go-micro/cmd"
"github.com/nats-io/go-nats"
2015-11-25 03:16:12 +03:00
)
type nbroker struct {
2018-11-05 09:39:28 +03:00
sync.RWMutex
2015-11-25 03:16:12 +03:00
addrs []string
conn *nats.Conn
2015-12-31 21:23:15 +03:00
opts broker.Options
2018-01-05 22:18:00 +03:00
nopts nats.Options
2015-11-25 03:16:12 +03:00
}
type subscriber struct {
s *nats.Subscription
opts broker.SubscribeOptions
}
type publication struct {
t string
m *broker.Message
2015-11-25 03:16:12 +03:00
}
func init() {
2016-01-02 22:25:20 +03:00
cmd.DefaultBrokers["nats"] = NewBroker
2015-11-25 03:16:12 +03:00
}
func (n *publication) Topic() string {
return n.t
}
func (n *publication) Message() *broker.Message {
return n.m
}
func (n *publication) Ack() error {
return nil
}
2015-12-31 21:23:15 +03:00
func (n *subscriber) Options() broker.SubscribeOptions {
return n.opts
}
2015-11-25 03:16:12 +03:00
func (n *subscriber) Topic() string {
return n.s.Subject
}
func (n *subscriber) Unsubscribe() error {
return n.s.Unsubscribe()
}
func (n *nbroker) Address() string {
if n.conn != nil && n.conn.IsConnected() {
return n.conn.ConnectedUrl()
}
2015-11-25 03:16:12 +03:00
if len(n.addrs) > 0 {
return n.addrs[0]
}
2015-11-25 03:16:12 +03:00
return ""
}
func setAddrs(addrs []string) []string {
var cAddrs []string
for _, addr := range addrs {
if len(addr) == 0 {
continue
}
if !strings.HasPrefix(addr, "nats://") {
addr = "nats://" + addr
}
cAddrs = append(cAddrs, addr)
}
if len(cAddrs) == 0 {
cAddrs = []string{nats.DefaultURL}
}
return cAddrs
}
2015-11-25 03:16:12 +03:00
func (n *nbroker) Connect() error {
2018-11-05 09:39:28 +03:00
n.RLock()
2018-09-03 02:15:45 +03:00
if n.conn != nil && n.conn.IsConnected() {
2018-11-05 09:39:28 +03:00
n.RUnlock()
2015-11-25 03:16:12 +03:00
return nil
}
2018-11-05 09:39:28 +03:00
n.RUnlock()
2015-11-25 03:16:12 +03:00
2018-01-05 22:18:00 +03:00
opts := n.nopts
2015-11-25 03:16:12 +03:00
opts.Servers = n.addrs
2016-01-17 02:11:20 +03:00
opts.Secure = n.opts.Secure
2016-01-17 02:44:55 +03:00
opts.TLSConfig = n.opts.TLSConfig
2016-01-17 02:11:20 +03:00
2016-01-17 03:37:55 +03:00
// secure might not be set
if n.opts.TLSConfig != nil {
opts.Secure = true
}
2015-11-25 03:16:12 +03:00
c, err := opts.Connect()
if err != nil {
return err
}
2018-11-05 09:39:28 +03:00
n.Lock()
2015-11-25 03:16:12 +03:00
n.conn = c
2018-11-05 09:39:28 +03:00
n.Unlock()
2015-11-25 03:16:12 +03:00
return nil
}
func (n *nbroker) Disconnect() error {
2018-11-05 09:39:28 +03:00
n.RLock()
2015-11-25 03:16:12 +03:00
n.conn.Close()
2018-11-05 09:39:28 +03:00
n.RUnlock()
2015-11-25 03:16:12 +03:00
return nil
}
func (n *nbroker) Init(opts ...broker.Option) error {
2016-01-17 02:11:20 +03:00
for _, o := range opts {
o(&n.opts)
}
n.addrs = setAddrs(n.opts.Addrs)
2015-11-25 03:16:12 +03:00
return nil
}
2015-12-31 21:23:15 +03:00
func (n *nbroker) Options() broker.Options {
return n.opts
}
func (n *nbroker) Publish(topic string, msg *broker.Message, opts ...broker.PublishOption) error {
b, err := n.opts.Codec.Marshal(msg)
2015-11-25 03:16:12 +03:00
if err != nil {
return err
}
2018-11-05 09:39:28 +03:00
n.RLock()
defer n.RUnlock()
2015-11-25 03:16:12 +03:00
return n.conn.Publish(topic, b)
}
func (n *nbroker) Subscribe(topic string, handler broker.Handler, opts ...broker.SubscribeOption) (broker.Subscriber, error) {
opt := broker.SubscribeOptions{
AutoAck: true,
}
for _, o := range opts {
o(&opt)
}
fn := func(msg *nats.Msg) {
2016-12-06 23:35:00 +03:00
var m broker.Message
if err := n.opts.Codec.Unmarshal(msg.Data, &m); err != nil {
2015-11-25 03:16:12 +03:00
return
}
2017-01-12 23:21:49 +03:00
handler(&publication{m: &m, t: msg.Subject})
}
var sub *nats.Subscription
var err error
2018-11-05 09:39:28 +03:00
n.RLock()
if len(opt.Queue) > 0 {
sub, err = n.conn.QueueSubscribe(topic, opt.Queue, fn)
} else {
sub, err = n.conn.Subscribe(topic, fn)
}
2018-11-05 09:39:28 +03:00
n.RUnlock()
2015-11-25 03:16:12 +03:00
if err != nil {
return nil, err
}
return &subscriber{s: sub, opts: opt}, nil
2015-11-25 03:16:12 +03:00
}
2015-12-20 00:56:42 +03:00
func (n *nbroker) String() string {
return "nats"
}
2016-03-16 13:08:10 +03:00
func NewBroker(opts ...broker.Option) broker.Broker {
options := broker.Options{
// Default codec
2019-01-10 12:42:14 +03:00
Codec: json.Marshaler{},
Context: context.Background(),
}
2016-03-16 13:08:10 +03:00
for _, o := range opts {
o(&options)
}
2018-01-06 18:19:15 +03:00
natsOpts := nats.GetDefaultOptions()
if n, ok := options.Context.Value(optionsKey{}).(nats.Options); ok {
2018-01-06 18:19:15 +03:00
natsOpts = n
}
2018-01-05 22:18:00 +03:00
// broker.Options have higher priority than nats.Options
// only if Addrs, Secure or TLSConfig were not set through a broker.Option
// we read them from nats.Option
if len(options.Addrs) == 0 {
2018-01-06 18:19:15 +03:00
options.Addrs = natsOpts.Servers
2018-01-05 22:18:00 +03:00
}
if !options.Secure {
2018-01-06 18:19:15 +03:00
options.Secure = natsOpts.Secure
2018-01-05 22:18:00 +03:00
}
if options.TLSConfig == nil {
2018-01-06 18:19:15 +03:00
options.TLSConfig = natsOpts.TLSConfig
2018-01-05 22:18:00 +03:00
}
nb := &nbroker{
2016-05-16 13:57:37 +03:00
opts: options,
2018-01-06 18:19:15 +03:00
nopts: natsOpts,
2018-01-05 22:18:00 +03:00
addrs: setAddrs(options.Addrs),
2015-11-25 03:16:12 +03:00
}
2018-01-05 22:18:00 +03:00
return nb
2015-11-25 03:16:12 +03:00
}