move implementations to external repos (#17)
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
This commit is contained in:
GitHub
@@ -16,11 +16,10 @@ type Broker interface {
|
||||
// Handler is used to process messages via a subscription of a topic.
|
||||
type Handler func(*Message) error
|
||||
|
||||
type ErrorHandler func(*Message, error)
|
||||
|
||||
type Message struct {
|
||||
Header map[string]string
|
||||
Body []byte
|
||||
Error error
|
||||
}
|
||||
|
||||
// Subscriber is a convenience return type for the Subscribe method
|
||||
|
||||
@@ -1,689 +0,0 @@
|
||||
// Package http provides a http based message broker
|
||||
package http
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"crypto/tls"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"io/ioutil"
|
||||
"math/rand"
|
||||
"net"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"runtime"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/google/uuid"
|
||||
"github.com/unistack-org/micro/v3/broker"
|
||||
"github.com/unistack-org/micro/v3/codec/json"
|
||||
merr "github.com/unistack-org/micro/v3/errors"
|
||||
"github.com/unistack-org/micro/v3/registry"
|
||||
"github.com/unistack-org/micro/v3/registry/cache"
|
||||
"github.com/unistack-org/micro/v3/registry/mdns"
|
||||
maddr "github.com/unistack-org/micro/v3/util/addr"
|
||||
mnet "github.com/unistack-org/micro/v3/util/net"
|
||||
mls "github.com/unistack-org/micro/v3/util/tls"
|
||||
"golang.org/x/net/http2"
|
||||
)
|
||||
|
||||
// HTTP Broker is a point to point async broker
|
||||
type httpBroker struct {
|
||||
id string
|
||||
address string
|
||||
opts broker.Options
|
||||
|
||||
mux *http.ServeMux
|
||||
|
||||
c *http.Client
|
||||
r registry.Registry
|
||||
|
||||
sync.RWMutex
|
||||
subscribers map[string][]*httpSubscriber
|
||||
running bool
|
||||
exit chan chan error
|
||||
|
||||
// offline message inbox
|
||||
mtx sync.RWMutex
|
||||
inbox map[string][][]byte
|
||||
}
|
||||
|
||||
type httpSubscriber struct {
|
||||
opts broker.SubscribeOptions
|
||||
id string
|
||||
topic string
|
||||
fn broker.Handler
|
||||
svc *registry.Service
|
||||
hb *httpBroker
|
||||
}
|
||||
|
||||
var (
|
||||
DefaultPath = "/"
|
||||
DefaultAddress = "127.0.0.1:0"
|
||||
serviceName = "micro.http.broker"
|
||||
broadcastVersion = "ff.http.broadcast"
|
||||
registerTTL = time.Minute
|
||||
registerInterval = time.Second * 30
|
||||
)
|
||||
|
||||
func init() {
|
||||
rand.Seed(time.Now().Unix())
|
||||
}
|
||||
|
||||
func newTransport(config *tls.Config) *http.Transport {
|
||||
if config == nil {
|
||||
config = &tls.Config{
|
||||
InsecureSkipVerify: true,
|
||||
}
|
||||
}
|
||||
|
||||
dialTLS := func(network string, addr string) (net.Conn, error) {
|
||||
return tls.Dial(network, addr, config)
|
||||
}
|
||||
|
||||
t := &http.Transport{
|
||||
Proxy: http.ProxyFromEnvironment,
|
||||
Dial: (&net.Dialer{
|
||||
Timeout: 30 * time.Second,
|
||||
KeepAlive: 30 * time.Second,
|
||||
}).Dial,
|
||||
TLSHandshakeTimeout: 10 * time.Second,
|
||||
DialTLS: dialTLS,
|
||||
}
|
||||
runtime.SetFinalizer(&t, func(tr **http.Transport) {
|
||||
(*tr).CloseIdleConnections()
|
||||
})
|
||||
|
||||
// setup http2
|
||||
http2.ConfigureTransport(t)
|
||||
|
||||
return t
|
||||
}
|
||||
|
||||
func newHttpBroker(opts ...broker.Option) broker.Broker {
|
||||
options := broker.Options{
|
||||
Codec: json.Marshaler{},
|
||||
Context: context.TODO(),
|
||||
Registry: mdns.NewRegistry(),
|
||||
}
|
||||
|
||||
for _, o := range opts {
|
||||
o(&options)
|
||||
}
|
||||
|
||||
// set address
|
||||
addr := DefaultAddress
|
||||
|
||||
if len(options.Addrs) > 0 && len(options.Addrs[0]) > 0 {
|
||||
addr = options.Addrs[0]
|
||||
}
|
||||
|
||||
h := &httpBroker{
|
||||
id: uuid.New().String(),
|
||||
address: addr,
|
||||
opts: options,
|
||||
r: options.Registry,
|
||||
c: &http.Client{Transport: newTransport(options.TLSConfig)},
|
||||
subscribers: make(map[string][]*httpSubscriber),
|
||||
exit: make(chan chan error),
|
||||
mux: http.NewServeMux(),
|
||||
inbox: make(map[string][][]byte),
|
||||
}
|
||||
|
||||
// specify the message handler
|
||||
h.mux.Handle(DefaultPath, h)
|
||||
|
||||
// get optional handlers
|
||||
if h.opts.Context != nil {
|
||||
handlers, ok := h.opts.Context.Value("http_handlers").(map[string]http.Handler)
|
||||
if ok {
|
||||
for pattern, handler := range handlers {
|
||||
h.mux.Handle(pattern, handler)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return h
|
||||
}
|
||||
|
||||
func (h *httpSubscriber) Options() broker.SubscribeOptions {
|
||||
return h.opts
|
||||
}
|
||||
|
||||
func (h *httpSubscriber) Topic() string {
|
||||
return h.topic
|
||||
}
|
||||
|
||||
func (h *httpSubscriber) Unsubscribe() error {
|
||||
return h.hb.unsubscribe(h)
|
||||
}
|
||||
|
||||
func (h *httpBroker) saveMessage(topic string, msg []byte) {
|
||||
h.mtx.Lock()
|
||||
defer h.mtx.Unlock()
|
||||
|
||||
// get messages
|
||||
c := h.inbox[topic]
|
||||
|
||||
// save message
|
||||
c = append(c, msg)
|
||||
|
||||
// max length 64
|
||||
if len(c) > 64 {
|
||||
c = c[:64]
|
||||
}
|
||||
|
||||
// save inbox
|
||||
h.inbox[topic] = c
|
||||
}
|
||||
|
||||
func (h *httpBroker) getMessage(topic string, num int) [][]byte {
|
||||
h.mtx.Lock()
|
||||
defer h.mtx.Unlock()
|
||||
|
||||
// get messages
|
||||
c, ok := h.inbox[topic]
|
||||
if !ok {
|
||||
return nil
|
||||
}
|
||||
|
||||
// more message than requests
|
||||
if len(c) >= num {
|
||||
msg := c[:num]
|
||||
h.inbox[topic] = c[num:]
|
||||
return msg
|
||||
}
|
||||
|
||||
// reset inbox
|
||||
h.inbox[topic] = nil
|
||||
|
||||
// return all messages
|
||||
return c
|
||||
}
|
||||
|
||||
func (h *httpBroker) subscribe(s *httpSubscriber) error {
|
||||
h.Lock()
|
||||
defer h.Unlock()
|
||||
|
||||
if err := h.r.Register(s.svc, registry.RegisterTTL(registerTTL)); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
h.subscribers[s.topic] = append(h.subscribers[s.topic], s)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (h *httpBroker) unsubscribe(s *httpSubscriber) error {
|
||||
h.Lock()
|
||||
defer h.Unlock()
|
||||
|
||||
//nolint:prealloc
|
||||
var subscribers []*httpSubscriber
|
||||
|
||||
// look for subscriber
|
||||
for _, sub := range h.subscribers[s.topic] {
|
||||
// deregister and skip forward
|
||||
if sub == s {
|
||||
_ = h.r.Deregister(sub.svc)
|
||||
continue
|
||||
}
|
||||
// keep subscriber
|
||||
subscribers = append(subscribers, sub)
|
||||
}
|
||||
|
||||
// set subscribers
|
||||
h.subscribers[s.topic] = subscribers
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (h *httpBroker) run(l net.Listener) {
|
||||
t := time.NewTicker(registerInterval)
|
||||
defer t.Stop()
|
||||
|
||||
for {
|
||||
select {
|
||||
// heartbeat for each subscriber
|
||||
case <-t.C:
|
||||
h.RLock()
|
||||
for _, subs := range h.subscribers {
|
||||
for _, sub := range subs {
|
||||
_ = h.r.Register(sub.svc, registry.RegisterTTL(registerTTL))
|
||||
}
|
||||
}
|
||||
h.RUnlock()
|
||||
// received exit signal
|
||||
case ch := <-h.exit:
|
||||
ch <- l.Close()
|
||||
h.RLock()
|
||||
for _, subs := range h.subscribers {
|
||||
for _, sub := range subs {
|
||||
_ = h.r.Deregister(sub.svc)
|
||||
}
|
||||
}
|
||||
h.RUnlock()
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (h *httpBroker) ServeHTTP(w http.ResponseWriter, req *http.Request) {
|
||||
if req.Method != "POST" {
|
||||
err := merr.BadRequest("go.micro.broker", "Method not allowed")
|
||||
http.Error(w, err.Error(), http.StatusMethodNotAllowed)
|
||||
return
|
||||
}
|
||||
defer req.Body.Close()
|
||||
|
||||
req.ParseForm()
|
||||
|
||||
b, err := ioutil.ReadAll(req.Body)
|
||||
if err != nil {
|
||||
errr := merr.InternalServerError("go.micro.broker", "Error reading request body: %v", err)
|
||||
w.WriteHeader(500)
|
||||
w.Write([]byte(errr.Error()))
|
||||
return
|
||||
}
|
||||
|
||||
var msg *broker.Message
|
||||
if err = h.opts.Codec.Unmarshal(b, &msg); err != nil {
|
||||
errr := merr.InternalServerError("go.micro.broker", "Error parsing request body: %v", err)
|
||||
w.WriteHeader(500)
|
||||
w.Write([]byte(errr.Error()))
|
||||
return
|
||||
}
|
||||
|
||||
topic := msg.Header["Micro-Topic"]
|
||||
|
||||
if len(topic) == 0 {
|
||||
errr := merr.InternalServerError("go.micro.broker", "Topic not found")
|
||||
w.WriteHeader(500)
|
||||
w.Write([]byte(errr.Error()))
|
||||
return
|
||||
}
|
||||
|
||||
id := req.Form.Get("id")
|
||||
|
||||
//nolint:prealloc
|
||||
var subs []broker.Handler
|
||||
|
||||
h.RLock()
|
||||
for _, subscriber := range h.subscribers[topic] {
|
||||
if id != subscriber.id {
|
||||
continue
|
||||
}
|
||||
subs = append(subs, subscriber.fn)
|
||||
}
|
||||
h.RUnlock()
|
||||
|
||||
// execute the handler
|
||||
for _, fn := range subs {
|
||||
fn(msg)
|
||||
}
|
||||
}
|
||||
|
||||
func (h *httpBroker) Address() string {
|
||||
h.RLock()
|
||||
defer h.RUnlock()
|
||||
return h.address
|
||||
}
|
||||
|
||||
func (h *httpBroker) Connect() error {
|
||||
h.RLock()
|
||||
if h.running {
|
||||
h.RUnlock()
|
||||
return nil
|
||||
}
|
||||
h.RUnlock()
|
||||
|
||||
h.Lock()
|
||||
defer h.Unlock()
|
||||
|
||||
var l net.Listener
|
||||
var err error
|
||||
|
||||
if h.opts.Secure || h.opts.TLSConfig != nil {
|
||||
config := h.opts.TLSConfig
|
||||
|
||||
fn := func(addr string) (net.Listener, error) {
|
||||
if config == nil {
|
||||
hosts := []string{addr}
|
||||
|
||||
// check if its a valid host:port
|
||||
if host, _, err := net.SplitHostPort(addr); err == nil {
|
||||
if len(host) == 0 {
|
||||
hosts = maddr.IPs()
|
||||
} else {
|
||||
hosts = []string{host}
|
||||
}
|
||||
}
|
||||
|
||||
// generate a certificate
|
||||
cert, err := mls.Certificate(hosts...)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
config = &tls.Config{Certificates: []tls.Certificate{cert}}
|
||||
}
|
||||
return tls.Listen("tcp", addr, config)
|
||||
}
|
||||
|
||||
l, err = mnet.Listen(h.address, fn)
|
||||
} else {
|
||||
fn := func(addr string) (net.Listener, error) {
|
||||
return net.Listen("tcp", addr)
|
||||
}
|
||||
|
||||
l, err = mnet.Listen(h.address, fn)
|
||||
}
|
||||
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
addr := h.address
|
||||
h.address = l.Addr().String()
|
||||
|
||||
go http.Serve(l, h.mux)
|
||||
go func() {
|
||||
h.run(l)
|
||||
h.Lock()
|
||||
h.opts.Addrs = []string{addr}
|
||||
h.address = addr
|
||||
h.Unlock()
|
||||
}()
|
||||
|
||||
// get registry
|
||||
reg := h.opts.Registry
|
||||
if reg == nil {
|
||||
reg = mdns.NewRegistry()
|
||||
}
|
||||
// set cache
|
||||
h.r = cache.New(reg)
|
||||
|
||||
// set running
|
||||
h.running = true
|
||||
return nil
|
||||
}
|
||||
|
||||
func (h *httpBroker) Disconnect() error {
|
||||
h.RLock()
|
||||
if !h.running {
|
||||
h.RUnlock()
|
||||
return nil
|
||||
}
|
||||
h.RUnlock()
|
||||
|
||||
h.Lock()
|
||||
defer h.Unlock()
|
||||
|
||||
// stop cache
|
||||
rc, ok := h.r.(cache.Cache)
|
||||
if ok {
|
||||
rc.Stop()
|
||||
}
|
||||
|
||||
// exit and return err
|
||||
ch := make(chan error)
|
||||
h.exit <- ch
|
||||
err := <-ch
|
||||
|
||||
// set not running
|
||||
h.running = false
|
||||
return err
|
||||
}
|
||||
|
||||
func (h *httpBroker) Init(opts ...broker.Option) error {
|
||||
h.RLock()
|
||||
if h.running {
|
||||
h.RUnlock()
|
||||
return errors.New("cannot init while connected")
|
||||
}
|
||||
h.RUnlock()
|
||||
|
||||
h.Lock()
|
||||
defer h.Unlock()
|
||||
|
||||
for _, o := range opts {
|
||||
o(&h.opts)
|
||||
}
|
||||
|
||||
if len(h.opts.Addrs) > 0 && len(h.opts.Addrs[0]) > 0 {
|
||||
h.address = h.opts.Addrs[0]
|
||||
}
|
||||
|
||||
if len(h.id) == 0 {
|
||||
h.id = "go.micro.http.broker-" + uuid.New().String()
|
||||
}
|
||||
|
||||
// get registry
|
||||
reg := h.opts.Registry
|
||||
if reg == nil {
|
||||
reg = mdns.NewRegistry()
|
||||
}
|
||||
|
||||
// get cache
|
||||
if rc, ok := h.r.(cache.Cache); ok {
|
||||
rc.Stop()
|
||||
}
|
||||
|
||||
// set registry
|
||||
h.r = cache.New(reg)
|
||||
|
||||
// reconfigure tls config
|
||||
if c := h.opts.TLSConfig; c != nil {
|
||||
h.c = &http.Client{
|
||||
Transport: newTransport(c),
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (h *httpBroker) Options() broker.Options {
|
||||
return h.opts
|
||||
}
|
||||
|
||||
func (h *httpBroker) Publish(topic string, msg *broker.Message, opts ...broker.PublishOption) error {
|
||||
// create the message first
|
||||
m := &broker.Message{
|
||||
Header: make(map[string]string),
|
||||
Body: msg.Body,
|
||||
}
|
||||
|
||||
for k, v := range msg.Header {
|
||||
m.Header[k] = v
|
||||
}
|
||||
|
||||
m.Header["Micro-Topic"] = topic
|
||||
|
||||
// encode the message
|
||||
b, err := h.opts.Codec.Marshal(m)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// save the message
|
||||
h.saveMessage(topic, b)
|
||||
|
||||
// now attempt to get the service
|
||||
h.RLock()
|
||||
s, err := h.r.GetService(serviceName)
|
||||
if err != nil {
|
||||
h.RUnlock()
|
||||
return err
|
||||
}
|
||||
h.RUnlock()
|
||||
|
||||
pub := func(node *registry.Node, t string, b []byte) error {
|
||||
scheme := "http"
|
||||
|
||||
// check if secure is added in metadata
|
||||
if node.Metadata["secure"] == "true" {
|
||||
scheme = "https"
|
||||
}
|
||||
|
||||
vals := url.Values{}
|
||||
vals.Add("id", node.Id)
|
||||
|
||||
uri := fmt.Sprintf("%s://%s%s?%s", scheme, node.Address, DefaultPath, vals.Encode())
|
||||
r, err := h.c.Post(uri, "application/json", bytes.NewReader(b))
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// discard response body
|
||||
io.Copy(ioutil.Discard, r.Body)
|
||||
r.Body.Close()
|
||||
return nil
|
||||
}
|
||||
|
||||
srv := func(s []*registry.Service, b []byte) {
|
||||
for _, service := range s {
|
||||
var nodes []*registry.Node
|
||||
|
||||
for _, node := range service.Nodes {
|
||||
// only use nodes tagged with broker http
|
||||
if node.Metadata["broker"] != "http" {
|
||||
continue
|
||||
}
|
||||
|
||||
// look for nodes for the topic
|
||||
if node.Metadata["topic"] != topic {
|
||||
continue
|
||||
}
|
||||
|
||||
nodes = append(nodes, node)
|
||||
}
|
||||
|
||||
// only process if we have nodes
|
||||
if len(nodes) == 0 {
|
||||
continue
|
||||
}
|
||||
|
||||
switch service.Version {
|
||||
// broadcast version means broadcast to all nodes
|
||||
case broadcastVersion:
|
||||
var success bool
|
||||
|
||||
// publish to all nodes
|
||||
for _, node := range nodes {
|
||||
// publish async
|
||||
if err := pub(node, topic, b); err == nil {
|
||||
success = true
|
||||
}
|
||||
}
|
||||
|
||||
// save if it failed to publish at least once
|
||||
if !success {
|
||||
h.saveMessage(topic, b)
|
||||
}
|
||||
default:
|
||||
// select node to publish to
|
||||
node := nodes[rand.Int()%len(nodes)]
|
||||
|
||||
// publish async to one node
|
||||
if err := pub(node, topic, b); err != nil {
|
||||
// if failed save it
|
||||
h.saveMessage(topic, b)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// do the rest async
|
||||
go func() {
|
||||
// get a third of the backlog
|
||||
messages := h.getMessage(topic, 8)
|
||||
delay := (len(messages) > 1)
|
||||
|
||||
// publish all the messages
|
||||
for _, msg := range messages {
|
||||
// serialize here
|
||||
srv(s, msg)
|
||||
|
||||
// sending a backlog of messages
|
||||
if delay {
|
||||
time.Sleep(time.Millisecond * 100)
|
||||
}
|
||||
}
|
||||
}()
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (h *httpBroker) Subscribe(topic string, handler broker.Handler, opts ...broker.SubscribeOption) (broker.Subscriber, error) {
|
||||
var err error
|
||||
var host, port string
|
||||
options := broker.NewSubscribeOptions(opts...)
|
||||
|
||||
// parse address for host, port
|
||||
host, port, err = net.SplitHostPort(h.Address())
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
addr, err := maddr.Extract(host)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
var secure bool
|
||||
|
||||
if h.opts.Secure || h.opts.TLSConfig != nil {
|
||||
secure = true
|
||||
}
|
||||
|
||||
// register service
|
||||
node := ®istry.Node{
|
||||
Id: topic + "-" + h.id,
|
||||
Address: mnet.HostPort(addr, port),
|
||||
Metadata: map[string]string{
|
||||
"secure": fmt.Sprintf("%t", secure),
|
||||
"broker": "http",
|
||||
"topic": topic,
|
||||
},
|
||||
}
|
||||
|
||||
// check for queue group or broadcast queue
|
||||
version := options.Queue
|
||||
if len(version) == 0 {
|
||||
version = broadcastVersion
|
||||
}
|
||||
|
||||
service := ®istry.Service{
|
||||
Name: serviceName,
|
||||
Version: version,
|
||||
Nodes: []*registry.Node{node},
|
||||
}
|
||||
|
||||
// generate subscriber
|
||||
subscriber := &httpSubscriber{
|
||||
opts: options,
|
||||
hb: h,
|
||||
id: node.Id,
|
||||
topic: topic,
|
||||
fn: handler,
|
||||
svc: service,
|
||||
}
|
||||
|
||||
// subscribe now
|
||||
if err := h.subscribe(subscriber); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
// return the subscriber
|
||||
return subscriber, nil
|
||||
}
|
||||
|
||||
func (h *httpBroker) String() string {
|
||||
return "http"
|
||||
}
|
||||
|
||||
// NewBroker returns a new http broker
|
||||
func NewBroker(opts ...broker.Option) broker.Broker {
|
||||
return newHttpBroker(opts...)
|
||||
}
|
||||
@@ -1,377 +0,0 @@
|
||||
package http
|
||||
|
||||
import (
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/google/uuid"
|
||||
"github.com/unistack-org/micro/v3/broker"
|
||||
"github.com/unistack-org/micro/v3/registry"
|
||||
"github.com/unistack-org/micro/v3/registry/memory"
|
||||
)
|
||||
|
||||
var (
|
||||
// mock data
|
||||
testData = map[string][]*registry.Service{
|
||||
"foo": {
|
||||
{
|
||||
Name: "foo",
|
||||
Version: "1.0.0",
|
||||
Nodes: []*registry.Node{
|
||||
{
|
||||
Id: "foo-1.0.0-123",
|
||||
Address: "localhost:9999",
|
||||
},
|
||||
{
|
||||
Id: "foo-1.0.0-321",
|
||||
Address: "localhost:9999",
|
||||
},
|
||||
},
|
||||
},
|
||||
{
|
||||
Name: "foo",
|
||||
Version: "1.0.1",
|
||||
Nodes: []*registry.Node{
|
||||
{
|
||||
Id: "foo-1.0.1-321",
|
||||
Address: "localhost:6666",
|
||||
},
|
||||
},
|
||||
},
|
||||
{
|
||||
Name: "foo",
|
||||
Version: "1.0.3",
|
||||
Nodes: []*registry.Node{
|
||||
{
|
||||
Id: "foo-1.0.3-345",
|
||||
Address: "localhost:8888",
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
)
|
||||
|
||||
func newTestRegistry() registry.Registry {
|
||||
return memory.NewRegistry(memory.Services(testData))
|
||||
}
|
||||
|
||||
func sub(be *testing.B, c int) {
|
||||
be.StopTimer()
|
||||
m := newTestRegistry()
|
||||
|
||||
b := NewBroker(broker.Registry(m))
|
||||
topic := uuid.New().String()
|
||||
|
||||
if err := b.Init(); err != nil {
|
||||
be.Fatalf("Unexpected init error: %v", err)
|
||||
}
|
||||
|
||||
if err := b.Connect(); err != nil {
|
||||
be.Fatalf("Unexpected connect error: %v", err)
|
||||
}
|
||||
|
||||
msg := &broker.Message{
|
||||
Header: map[string]string{
|
||||
"Content-Type": "application/json",
|
||||
},
|
||||
Body: []byte(`{"message": "Hello World"}`),
|
||||
}
|
||||
|
||||
var subs []broker.Subscriber
|
||||
done := make(chan bool, c)
|
||||
|
||||
for i := 0; i < c; i++ {
|
||||
sub, err := b.Subscribe(topic, func(m *broker.Message) error {
|
||||
done <- true
|
||||
|
||||
if string(m.Body) != string(msg.Body) {
|
||||
be.Fatalf("Unexpected msg %s, expected %s", string(m.Body), string(msg.Body))
|
||||
}
|
||||
|
||||
return nil
|
||||
}, broker.Queue("shared"))
|
||||
if err != nil {
|
||||
be.Fatalf("Unexpected subscribe error: %v", err)
|
||||
}
|
||||
subs = append(subs, sub)
|
||||
}
|
||||
|
||||
for i := 0; i < be.N; i++ {
|
||||
be.StartTimer()
|
||||
if err := b.Publish(topic, msg); err != nil {
|
||||
be.Fatalf("Unexpected publish error: %v", err)
|
||||
}
|
||||
<-done
|
||||
be.StopTimer()
|
||||
}
|
||||
|
||||
for _, sub := range subs {
|
||||
sub.Unsubscribe()
|
||||
}
|
||||
|
||||
if err := b.Disconnect(); err != nil {
|
||||
be.Fatalf("Unexpected disconnect error: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func pub(be *testing.B, c int) {
|
||||
be.StopTimer()
|
||||
m := newTestRegistry()
|
||||
b := NewBroker(broker.Registry(m))
|
||||
topic := uuid.New().String()
|
||||
|
||||
if err := b.Init(); err != nil {
|
||||
be.Fatalf("Unexpected init error: %v", err)
|
||||
}
|
||||
|
||||
if err := b.Connect(); err != nil {
|
||||
be.Fatalf("Unexpected connect error: %v", err)
|
||||
}
|
||||
|
||||
msg := &broker.Message{
|
||||
Header: map[string]string{
|
||||
"Content-Type": "application/json",
|
||||
},
|
||||
Body: []byte(`{"message": "Hello World"}`),
|
||||
}
|
||||
|
||||
done := make(chan bool, c*4)
|
||||
|
||||
sub, err := b.Subscribe(topic, func(m *broker.Message) error {
|
||||
done <- true
|
||||
if string(m.Body) != string(msg.Body) {
|
||||
be.Fatalf("Unexpected msg %s, expected %s", string(m.Body), string(msg.Body))
|
||||
}
|
||||
return nil
|
||||
}, broker.Queue("shared"))
|
||||
if err != nil {
|
||||
be.Fatalf("Unexpected subscribe error: %v", err)
|
||||
}
|
||||
|
||||
var wg sync.WaitGroup
|
||||
ch := make(chan int, c*4)
|
||||
be.StartTimer()
|
||||
|
||||
for i := 0; i < c; i++ {
|
||||
go func() {
|
||||
for range ch {
|
||||
if err := b.Publish(topic, msg); err != nil {
|
||||
be.Fatalf("Unexpected publish error: %v", err)
|
||||
}
|
||||
select {
|
||||
case <-done:
|
||||
case <-time.After(time.Second):
|
||||
}
|
||||
wg.Done()
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
for i := 0; i < be.N; i++ {
|
||||
wg.Add(1)
|
||||
ch <- i
|
||||
}
|
||||
|
||||
wg.Wait()
|
||||
be.StopTimer()
|
||||
sub.Unsubscribe()
|
||||
close(ch)
|
||||
close(done)
|
||||
|
||||
if err := b.Disconnect(); err != nil {
|
||||
be.Fatalf("Unexpected disconnect error: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestBroker(t *testing.T) {
|
||||
m := newTestRegistry()
|
||||
b := NewBroker(broker.Registry(m))
|
||||
|
||||
if err := b.Init(); err != nil {
|
||||
t.Fatalf("Unexpected init error: %v", err)
|
||||
}
|
||||
|
||||
if err := b.Connect(); err != nil {
|
||||
t.Fatalf("Unexpected connect error: %v", err)
|
||||
}
|
||||
|
||||
msg := &broker.Message{
|
||||
Header: map[string]string{
|
||||
"Content-Type": "application/json",
|
||||
},
|
||||
Body: []byte(`{"message": "Hello World"}`),
|
||||
}
|
||||
|
||||
done := make(chan bool)
|
||||
|
||||
sub, err := b.Subscribe("test", func(m *broker.Message) error {
|
||||
|
||||
if string(m.Body) != string(msg.Body) {
|
||||
t.Fatalf("Unexpected msg %s, expected %s", string(m.Body), string(msg.Body))
|
||||
}
|
||||
|
||||
close(done)
|
||||
return nil
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("Unexpected subscribe error: %v", err)
|
||||
}
|
||||
|
||||
if err := b.Publish("test", msg); err != nil {
|
||||
t.Fatalf("Unexpected publish error: %v", err)
|
||||
}
|
||||
|
||||
<-done
|
||||
sub.Unsubscribe()
|
||||
|
||||
if err := b.Disconnect(); err != nil {
|
||||
t.Fatalf("Unexpected disconnect error: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestConcurrentSubBroker(t *testing.T) {
|
||||
m := newTestRegistry()
|
||||
b := NewBroker(broker.Registry(m))
|
||||
|
||||
if err := b.Init(); err != nil {
|
||||
t.Fatalf("Unexpected init error: %v", err)
|
||||
}
|
||||
|
||||
if err := b.Connect(); err != nil {
|
||||
t.Fatalf("Unexpected connect error: %v", err)
|
||||
}
|
||||
|
||||
msg := &broker.Message{
|
||||
Header: map[string]string{
|
||||
"Content-Type": "application/json",
|
||||
},
|
||||
Body: []byte(`{"message": "Hello World"}`),
|
||||
}
|
||||
|
||||
var subs []broker.Subscriber
|
||||
var wg sync.WaitGroup
|
||||
|
||||
for i := 0; i < 10; i++ {
|
||||
sub, err := b.Subscribe("test", func(m *broker.Message) error {
|
||||
defer wg.Done()
|
||||
|
||||
if string(m.Body) != string(msg.Body) {
|
||||
t.Fatalf("Unexpected msg %s, expected %s", string(m.Body), string(msg.Body))
|
||||
}
|
||||
|
||||
return nil
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("Unexpected subscribe error: %v", err)
|
||||
}
|
||||
|
||||
wg.Add(1)
|
||||
subs = append(subs, sub)
|
||||
}
|
||||
|
||||
if err := b.Publish("test", msg); err != nil {
|
||||
t.Fatalf("Unexpected publish error: %v", err)
|
||||
}
|
||||
|
||||
wg.Wait()
|
||||
|
||||
for _, sub := range subs {
|
||||
sub.Unsubscribe()
|
||||
}
|
||||
|
||||
if err := b.Disconnect(); err != nil {
|
||||
t.Fatalf("Unexpected disconnect error: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestConcurrentPubBroker(t *testing.T) {
|
||||
m := newTestRegistry()
|
||||
b := NewBroker(broker.Registry(m))
|
||||
|
||||
if err := b.Init(); err != nil {
|
||||
t.Fatalf("Unexpected init error: %v", err)
|
||||
}
|
||||
|
||||
if err := b.Connect(); err != nil {
|
||||
t.Fatalf("Unexpected connect error: %v", err)
|
||||
}
|
||||
|
||||
msg := &broker.Message{
|
||||
Header: map[string]string{
|
||||
"Content-Type": "application/json",
|
||||
},
|
||||
Body: []byte(`{"message": "Hello World"}`),
|
||||
}
|
||||
|
||||
var wg sync.WaitGroup
|
||||
|
||||
sub, err := b.Subscribe("test", func(m *broker.Message) error {
|
||||
defer wg.Done()
|
||||
|
||||
if string(m.Body) != string(msg.Body) {
|
||||
t.Fatalf("Unexpected msg %s, expected %s", string(m.Body), string(msg.Body))
|
||||
}
|
||||
|
||||
return nil
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("Unexpected subscribe error: %v", err)
|
||||
}
|
||||
|
||||
for i := 0; i < 10; i++ {
|
||||
wg.Add(1)
|
||||
|
||||
if err := b.Publish("test", msg); err != nil {
|
||||
t.Fatalf("Unexpected publish error: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
wg.Wait()
|
||||
|
||||
sub.Unsubscribe()
|
||||
|
||||
if err := b.Disconnect(); err != nil {
|
||||
t.Fatalf("Unexpected disconnect error: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func BenchmarkSub1(b *testing.B) {
|
||||
sub(b, 1)
|
||||
}
|
||||
func BenchmarkSub8(b *testing.B) {
|
||||
sub(b, 8)
|
||||
}
|
||||
|
||||
func BenchmarkSub32(b *testing.B) {
|
||||
sub(b, 32)
|
||||
}
|
||||
|
||||
func BenchmarkSub64(b *testing.B) {
|
||||
sub(b, 64)
|
||||
}
|
||||
|
||||
func BenchmarkSub128(b *testing.B) {
|
||||
sub(b, 128)
|
||||
}
|
||||
|
||||
func BenchmarkPub1(b *testing.B) {
|
||||
pub(b, 1)
|
||||
}
|
||||
|
||||
func BenchmarkPub8(b *testing.B) {
|
||||
pub(b, 8)
|
||||
}
|
||||
|
||||
func BenchmarkPub32(b *testing.B) {
|
||||
pub(b, 32)
|
||||
}
|
||||
|
||||
func BenchmarkPub64(b *testing.B) {
|
||||
pub(b, 64)
|
||||
}
|
||||
|
||||
func BenchmarkPub128(b *testing.B) {
|
||||
pub(b, 128)
|
||||
}
|
||||
@@ -1,23 +0,0 @@
|
||||
package http
|
||||
|
||||
import (
|
||||
"context"
|
||||
"net/http"
|
||||
|
||||
"github.com/unistack-org/micro/v3/broker"
|
||||
)
|
||||
|
||||
// Handle registers the handler for the given pattern.
|
||||
func Handle(pattern string, handler http.Handler) broker.Option {
|
||||
return func(o *broker.Options) {
|
||||
if o.Context == nil {
|
||||
o.Context = context.Background()
|
||||
}
|
||||
handlers, ok := o.Context.Value("http_handlers").(map[string]http.Handler)
|
||||
if !ok {
|
||||
handlers = make(map[string]http.Handler)
|
||||
}
|
||||
handlers[pattern] = handler
|
||||
o.Context = context.WithValue(o.Context, "http_handlers", handlers)
|
||||
}
|
||||
}
|
||||
@@ -1,183 +0,0 @@
|
||||
// Package memory provides a memory broker
|
||||
package memory
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"math/rand"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/google/uuid"
|
||||
"github.com/unistack-org/micro/v3/broker"
|
||||
maddr "github.com/unistack-org/micro/v3/util/addr"
|
||||
mnet "github.com/unistack-org/micro/v3/util/net"
|
||||
)
|
||||
|
||||
type memoryBroker struct {
|
||||
opts broker.Options
|
||||
|
||||
addr string
|
||||
sync.RWMutex
|
||||
connected bool
|
||||
Subscribers map[string][]*memorySubscriber
|
||||
}
|
||||
|
||||
type memorySubscriber struct {
|
||||
id string
|
||||
topic string
|
||||
exit chan bool
|
||||
handler broker.Handler
|
||||
opts broker.SubscribeOptions
|
||||
}
|
||||
|
||||
func (m *memoryBroker) Options() broker.Options {
|
||||
return m.opts
|
||||
}
|
||||
|
||||
func (m *memoryBroker) Address() string {
|
||||
return m.addr
|
||||
}
|
||||
|
||||
func (m *memoryBroker) Connect() error {
|
||||
m.Lock()
|
||||
defer m.Unlock()
|
||||
|
||||
if m.connected {
|
||||
return nil
|
||||
}
|
||||
|
||||
// use 127.0.0.1 to avoid scan of all network interfaces
|
||||
addr, err := maddr.Extract("127.0.0.1")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
i := rand.Intn(20000)
|
||||
// set addr with port
|
||||
addr = mnet.HostPort(addr, 10000+i)
|
||||
|
||||
m.addr = addr
|
||||
m.connected = true
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (m *memoryBroker) Disconnect() error {
|
||||
m.Lock()
|
||||
defer m.Unlock()
|
||||
|
||||
if !m.connected {
|
||||
return nil
|
||||
}
|
||||
|
||||
m.connected = false
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (m *memoryBroker) Init(opts ...broker.Option) error {
|
||||
for _, o := range opts {
|
||||
o(&m.opts)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (m *memoryBroker) Publish(topic string, msg *broker.Message, opts ...broker.PublishOption) error {
|
||||
m.RLock()
|
||||
if !m.connected {
|
||||
m.RUnlock()
|
||||
return errors.New("not connected")
|
||||
}
|
||||
|
||||
subs, ok := m.Subscribers[topic]
|
||||
m.RUnlock()
|
||||
if !ok {
|
||||
return nil
|
||||
}
|
||||
|
||||
for _, sub := range subs {
|
||||
if err := sub.handler(msg); err != nil {
|
||||
if eh := sub.opts.ErrorHandler; eh != nil {
|
||||
eh(msg, err)
|
||||
}
|
||||
continue
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (m *memoryBroker) Subscribe(topic string, handler broker.Handler, opts ...broker.SubscribeOption) (broker.Subscriber, error) {
|
||||
m.RLock()
|
||||
if !m.connected {
|
||||
m.RUnlock()
|
||||
return nil, errors.New("not connected")
|
||||
}
|
||||
m.RUnlock()
|
||||
|
||||
var options broker.SubscribeOptions
|
||||
for _, o := range opts {
|
||||
o(&options)
|
||||
}
|
||||
|
||||
sub := &memorySubscriber{
|
||||
exit: make(chan bool, 1),
|
||||
id: uuid.New().String(),
|
||||
topic: topic,
|
||||
handler: handler,
|
||||
opts: options,
|
||||
}
|
||||
|
||||
m.Lock()
|
||||
m.Subscribers[topic] = append(m.Subscribers[topic], sub)
|
||||
m.Unlock()
|
||||
|
||||
go func() {
|
||||
<-sub.exit
|
||||
m.Lock()
|
||||
var newSubscribers []*memorySubscriber
|
||||
for _, sb := range m.Subscribers[topic] {
|
||||
if sb.id == sub.id {
|
||||
continue
|
||||
}
|
||||
newSubscribers = append(newSubscribers, sb)
|
||||
}
|
||||
m.Subscribers[topic] = newSubscribers
|
||||
m.Unlock()
|
||||
}()
|
||||
|
||||
return sub, nil
|
||||
}
|
||||
|
||||
func (m *memoryBroker) String() string {
|
||||
return "memory"
|
||||
}
|
||||
|
||||
func (m *memorySubscriber) Options() broker.SubscribeOptions {
|
||||
return m.opts
|
||||
}
|
||||
|
||||
func (m *memorySubscriber) Topic() string {
|
||||
return m.topic
|
||||
}
|
||||
|
||||
func (m *memorySubscriber) Unsubscribe() error {
|
||||
m.exit <- true
|
||||
return nil
|
||||
}
|
||||
|
||||
func NewBroker(opts ...broker.Option) broker.Broker {
|
||||
options := broker.Options{
|
||||
Context: context.Background(),
|
||||
}
|
||||
|
||||
rand.Seed(time.Now().UnixNano())
|
||||
for _, o := range opts {
|
||||
o(&options)
|
||||
}
|
||||
|
||||
return &memoryBroker{
|
||||
opts: options,
|
||||
Subscribers: make(map[string][]*memorySubscriber),
|
||||
}
|
||||
}
|
||||
@@ -1,50 +0,0 @@
|
||||
package memory
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"testing"
|
||||
|
||||
"github.com/unistack-org/micro/v3/broker"
|
||||
)
|
||||
|
||||
func TestMemoryBroker(t *testing.T) {
|
||||
b := NewBroker()
|
||||
|
||||
if err := b.Connect(); err != nil {
|
||||
t.Fatalf("Unexpected connect error %v", err)
|
||||
}
|
||||
|
||||
topic := "test"
|
||||
count := 10
|
||||
|
||||
fn := func(m *broker.Message) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
sub, err := b.Subscribe(topic, fn)
|
||||
if err != nil {
|
||||
t.Fatalf("Unexpected error subscribing %v", err)
|
||||
}
|
||||
|
||||
for i := 0; i < count; i++ {
|
||||
message := &broker.Message{
|
||||
Header: map[string]string{
|
||||
"foo": "bar",
|
||||
"id": fmt.Sprintf("%d", i),
|
||||
},
|
||||
Body: []byte(`hello world`),
|
||||
}
|
||||
|
||||
if err := b.Publish(topic, message); err != nil {
|
||||
t.Fatalf("Unexpected error publishing %d", i)
|
||||
}
|
||||
}
|
||||
|
||||
if err := sub.Unsubscribe(); err != nil {
|
||||
t.Fatalf("Unexpected error unsubscribing from %s: %v", topic, err)
|
||||
}
|
||||
|
||||
if err := b.Disconnect(); err != nil {
|
||||
t.Fatalf("Unexpected connect error %v", err)
|
||||
}
|
||||
}
|
||||
@@ -1,17 +0,0 @@
|
||||
package nats
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/unistack-org/micro/v3/broker"
|
||||
)
|
||||
|
||||
// setBrokerOption returns a function to setup a context with given value
|
||||
func setBrokerOption(k, v interface{}) broker.Option {
|
||||
return func(o *broker.Options) {
|
||||
if o.Context == nil {
|
||||
o.Context = context.Background()
|
||||
}
|
||||
o.Context = context.WithValue(o.Context, k, v)
|
||||
}
|
||||
}
|
||||
@@ -1,294 +0,0 @@
|
||||
// Package nats provides a NATS broker
|
||||
package nats
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"strings"
|
||||
"sync"
|
||||
|
||||
nats "github.com/nats-io/nats.go"
|
||||
"github.com/unistack-org/micro/v3/broker"
|
||||
"github.com/unistack-org/micro/v3/codec/json"
|
||||
"github.com/unistack-org/micro/v3/logger"
|
||||
"github.com/unistack-org/micro/v3/registry/mdns"
|
||||
)
|
||||
|
||||
type natsBroker struct {
|
||||
sync.Once
|
||||
sync.RWMutex
|
||||
|
||||
// indicate if we're connected
|
||||
connected bool
|
||||
|
||||
addrs []string
|
||||
conn *nats.Conn
|
||||
opts broker.Options
|
||||
nopts nats.Options
|
||||
|
||||
// should we drain the connection
|
||||
drain bool
|
||||
closeCh chan (error)
|
||||
}
|
||||
|
||||
type subscriber struct {
|
||||
s *nats.Subscription
|
||||
opts broker.SubscribeOptions
|
||||
}
|
||||
|
||||
func (s *subscriber) Options() broker.SubscribeOptions {
|
||||
return s.opts
|
||||
}
|
||||
|
||||
func (s *subscriber) Topic() string {
|
||||
return s.s.Subject
|
||||
}
|
||||
|
||||
func (s *subscriber) Unsubscribe() error {
|
||||
return s.s.Unsubscribe()
|
||||
}
|
||||
|
||||
func (n *natsBroker) Address() string {
|
||||
if n.conn != nil && n.conn.IsConnected() {
|
||||
return n.conn.ConnectedUrl()
|
||||
}
|
||||
|
||||
if len(n.addrs) > 0 {
|
||||
return n.addrs[0]
|
||||
}
|
||||
|
||||
return ""
|
||||
}
|
||||
|
||||
func (n *natsBroker) setAddrs(addrs []string) []string {
|
||||
//nolint:prealloc
|
||||
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
|
||||
}
|
||||
|
||||
func (n *natsBroker) Connect() error {
|
||||
n.Lock()
|
||||
defer n.Unlock()
|
||||
|
||||
if n.connected {
|
||||
return nil
|
||||
}
|
||||
|
||||
status := nats.CLOSED
|
||||
if n.conn != nil {
|
||||
status = n.conn.Status()
|
||||
}
|
||||
|
||||
switch status {
|
||||
case nats.CONNECTED, nats.RECONNECTING, nats.CONNECTING:
|
||||
n.connected = true
|
||||
return nil
|
||||
default: // DISCONNECTED or CLOSED or DRAINING
|
||||
opts := n.nopts
|
||||
opts.Servers = n.addrs
|
||||
opts.Secure = n.opts.Secure
|
||||
opts.TLSConfig = n.opts.TLSConfig
|
||||
|
||||
// secure might not be set
|
||||
if n.opts.TLSConfig != nil {
|
||||
opts.Secure = true
|
||||
}
|
||||
|
||||
c, err := opts.Connect()
|
||||
if err != nil {
|
||||
if logger.V(logger.WarnLevel, logger.DefaultLogger) {
|
||||
logger.Warnf("Error connecting to broker: %v", err)
|
||||
}
|
||||
|
||||
return err
|
||||
}
|
||||
n.conn = c
|
||||
n.connected = true
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
||||
func (n *natsBroker) Disconnect() error {
|
||||
n.Lock()
|
||||
defer n.Unlock()
|
||||
|
||||
// drain the connection if specified
|
||||
if n.drain {
|
||||
n.conn.Drain()
|
||||
n.closeCh <- nil
|
||||
}
|
||||
|
||||
// close the client connection
|
||||
n.conn.Close()
|
||||
|
||||
// set not connected
|
||||
n.connected = false
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (n *natsBroker) Init(opts ...broker.Option) error {
|
||||
n.setOption(opts...)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (n *natsBroker) Options() broker.Options {
|
||||
return n.opts
|
||||
}
|
||||
|
||||
func (n *natsBroker) Publish(topic string, msg *broker.Message, opts ...broker.PublishOption) error {
|
||||
n.RLock()
|
||||
defer n.RUnlock()
|
||||
|
||||
if n.conn == nil {
|
||||
return errors.New("not connected")
|
||||
}
|
||||
|
||||
b, err := n.opts.Codec.Marshal(msg)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return n.conn.Publish(topic, b)
|
||||
}
|
||||
|
||||
func (n *natsBroker) Subscribe(topic string, handler broker.Handler, opts ...broker.SubscribeOption) (broker.Subscriber, error) {
|
||||
n.RLock()
|
||||
if n.conn == nil {
|
||||
n.RUnlock()
|
||||
return nil, errors.New("not connected")
|
||||
}
|
||||
n.RUnlock()
|
||||
|
||||
opt := broker.SubscribeOptions{
|
||||
Context: context.Background(),
|
||||
}
|
||||
|
||||
for _, o := range opts {
|
||||
o(&opt)
|
||||
}
|
||||
|
||||
fn := func(msg *nats.Msg) {
|
||||
var m *broker.Message
|
||||
eh := opt.ErrorHandler
|
||||
err := n.opts.Codec.Unmarshal(msg.Data, &m)
|
||||
if err != nil {
|
||||
m.Body = msg.Data
|
||||
if logger.V(logger.ErrorLevel, logger.DefaultLogger) {
|
||||
logger.Error(err)
|
||||
}
|
||||
if eh != nil {
|
||||
eh(m, err)
|
||||
}
|
||||
return
|
||||
}
|
||||
if err := handler(m); err != nil {
|
||||
if logger.V(logger.ErrorLevel, logger.DefaultLogger) {
|
||||
logger.Error(err)
|
||||
}
|
||||
if eh != nil {
|
||||
eh(m, err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
var sub *nats.Subscription
|
||||
var err error
|
||||
|
||||
n.RLock()
|
||||
if len(opt.Queue) > 0 {
|
||||
sub, err = n.conn.QueueSubscribe(topic, opt.Queue, fn)
|
||||
} else {
|
||||
sub, err = n.conn.Subscribe(topic, fn)
|
||||
}
|
||||
n.RUnlock()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &subscriber{s: sub, opts: opt}, nil
|
||||
}
|
||||
|
||||
func (n *natsBroker) String() string {
|
||||
return "nats"
|
||||
}
|
||||
|
||||
func (n *natsBroker) setOption(opts ...broker.Option) {
|
||||
for _, o := range opts {
|
||||
o(&n.opts)
|
||||
}
|
||||
|
||||
n.Once.Do(func() {
|
||||
n.nopts = nats.GetDefaultOptions()
|
||||
})
|
||||
|
||||
if nopts, ok := n.opts.Context.Value(optionsKey{}).(nats.Options); ok {
|
||||
n.nopts = nopts
|
||||
}
|
||||
|
||||
// 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(n.opts.Addrs) == 0 {
|
||||
n.opts.Addrs = n.nopts.Servers
|
||||
}
|
||||
|
||||
if !n.opts.Secure {
|
||||
n.opts.Secure = n.nopts.Secure
|
||||
}
|
||||
|
||||
if n.opts.TLSConfig == nil {
|
||||
n.opts.TLSConfig = n.nopts.TLSConfig
|
||||
}
|
||||
n.addrs = n.setAddrs(n.opts.Addrs)
|
||||
|
||||
if n.opts.Context.Value(drainConnectionKey{}) != nil {
|
||||
n.drain = true
|
||||
n.closeCh = make(chan error)
|
||||
n.nopts.ClosedCB = n.onClose
|
||||
n.nopts.AsyncErrorCB = n.onAsyncError
|
||||
n.nopts.DisconnectedErrCB = n.onDisconnectedError
|
||||
}
|
||||
}
|
||||
|
||||
func (n *natsBroker) onClose(conn *nats.Conn) {
|
||||
n.closeCh <- nil
|
||||
}
|
||||
|
||||
func (n *natsBroker) onAsyncError(conn *nats.Conn, sub *nats.Subscription, err error) {
|
||||
// There are kinds of different async error nats might callback, but we are interested
|
||||
// in ErrDrainTimeout only here.
|
||||
if err == nats.ErrDrainTimeout {
|
||||
n.closeCh <- err
|
||||
}
|
||||
}
|
||||
|
||||
func (n *natsBroker) onDisconnectedError(conn *nats.Conn, err error) {
|
||||
n.closeCh <- err
|
||||
}
|
||||
|
||||
func NewBroker(opts ...broker.Option) broker.Broker {
|
||||
options := broker.Options{
|
||||
// Default codec
|
||||
Codec: json.Marshaler{},
|
||||
Context: context.Background(),
|
||||
Registry: mdns.NewRegistry(),
|
||||
}
|
||||
|
||||
n := &natsBroker{
|
||||
opts: options,
|
||||
}
|
||||
n.setOption(opts...)
|
||||
|
||||
return n
|
||||
}
|
||||
@@ -1,98 +0,0 @@
|
||||
package nats
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"testing"
|
||||
|
||||
nats "github.com/nats-io/nats.go"
|
||||
"github.com/unistack-org/micro/v3/broker"
|
||||
)
|
||||
|
||||
var addrTestCases = []struct {
|
||||
name string
|
||||
description string
|
||||
addrs map[string]string // expected address : set address
|
||||
}{
|
||||
{
|
||||
"brokerOpts",
|
||||
"set broker addresses through a broker.Option in constructor",
|
||||
map[string]string{
|
||||
"nats://192.168.10.1:5222": "192.168.10.1:5222",
|
||||
"nats://10.20.10.0:4222": "10.20.10.0:4222"},
|
||||
},
|
||||
{
|
||||
"brokerInit",
|
||||
"set broker addresses through a broker.Option in broker.Init()",
|
||||
map[string]string{
|
||||
"nats://192.168.10.1:5222": "192.168.10.1:5222",
|
||||
"nats://10.20.10.0:4222": "10.20.10.0:4222"},
|
||||
},
|
||||
{
|
||||
"natsOpts",
|
||||
"set broker addresses through the nats.Option in constructor",
|
||||
map[string]string{
|
||||
"nats://192.168.10.1:5222": "192.168.10.1:5222",
|
||||
"nats://10.20.10.0:4222": "10.20.10.0:4222"},
|
||||
},
|
||||
{
|
||||
"default",
|
||||
"check if default Address is set correctly",
|
||||
map[string]string{
|
||||
"nats://127.0.0.1:4222": "",
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
// TestInitAddrs tests issue #100. Ensures that if the addrs is set by an option in init it will be used.
|
||||
func TestInitAddrs(t *testing.T) {
|
||||
|
||||
for _, tc := range addrTestCases {
|
||||
t.Run(fmt.Sprintf("%s: %s", tc.name, tc.description), func(t *testing.T) {
|
||||
|
||||
var br broker.Broker
|
||||
var addrs []string
|
||||
|
||||
for _, addr := range tc.addrs {
|
||||
addrs = append(addrs, addr)
|
||||
}
|
||||
|
||||
switch tc.name {
|
||||
case "brokerOpts":
|
||||
// we know that there are just two addrs in the dict
|
||||
br = NewBroker(broker.Addrs(addrs[0], addrs[1]))
|
||||
br.Init()
|
||||
case "brokerInit":
|
||||
br = NewBroker()
|
||||
// we know that there are just two addrs in the dict
|
||||
br.Init(broker.Addrs(addrs[0], addrs[1]))
|
||||
case "natsOpts":
|
||||
nopts := nats.GetDefaultOptions()
|
||||
nopts.Servers = addrs
|
||||
br = NewBroker(Options(nopts))
|
||||
br.Init()
|
||||
case "default":
|
||||
br = NewBroker()
|
||||
br.Init()
|
||||
}
|
||||
|
||||
natsBroker, ok := br.(*natsBroker)
|
||||
if !ok {
|
||||
t.Fatal("Expected broker to be of types *natsBroker")
|
||||
}
|
||||
// check if the same amount of addrs we set has actually been set, default
|
||||
// have only 1 address nats://127.0.0.1:4222 (current nats code) or
|
||||
// nats://localhost:4222 (older code version)
|
||||
if len(natsBroker.addrs) != len(tc.addrs) && tc.name != "default" {
|
||||
t.Errorf("Expected Addr count = %d, Actual Addr count = %d",
|
||||
len(natsBroker.addrs), len(tc.addrs))
|
||||
}
|
||||
|
||||
for _, addr := range natsBroker.addrs {
|
||||
_, ok := tc.addrs[addr]
|
||||
if !ok {
|
||||
t.Errorf("Expected '%s' has not been set", addr)
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -1,19 +0,0 @@
|
||||
package nats
|
||||
|
||||
import (
|
||||
nats "github.com/nats-io/nats.go"
|
||||
"github.com/unistack-org/micro/v3/broker"
|
||||
)
|
||||
|
||||
type optionsKey struct{}
|
||||
type drainConnectionKey struct{}
|
||||
|
||||
// Options accepts nats.Options
|
||||
func Options(opts nats.Options) broker.Option {
|
||||
return setBrokerOption(optionsKey{}, opts)
|
||||
}
|
||||
|
||||
// DrainConnection will drain subscription on close
|
||||
func DrainConnection() broker.Option {
|
||||
return setBrokerOption(drainConnectionKey{}, struct{}{})
|
||||
}
|
||||
@@ -9,10 +9,15 @@ import (
|
||||
)
|
||||
|
||||
type Options struct {
|
||||
AutoAck bool
|
||||
|
||||
Addrs []string
|
||||
Secure bool
|
||||
Codec codec.Marshaler
|
||||
|
||||
// Handler executed when errors occur processing messages
|
||||
ErrorHandler Handler
|
||||
|
||||
TLSConfig *tls.Config
|
||||
// Registry used for clustering
|
||||
Registry registry.Registry
|
||||
@@ -28,13 +33,16 @@ type PublishOptions struct {
|
||||
}
|
||||
|
||||
type SubscribeOptions struct {
|
||||
// Handler executed when errors occur processing messages
|
||||
ErrorHandler ErrorHandler
|
||||
// AutoAck ack messages if handler returns nil err
|
||||
AutoAck bool
|
||||
|
||||
// Subscribers with the same queue name
|
||||
// Handler executed when errors occur processing messages
|
||||
ErrorHandler Handler
|
||||
|
||||
// Subscribers with the same group name
|
||||
// will create a shared subscription where each
|
||||
// receives a subset of messages.
|
||||
Queue string
|
||||
Group string
|
||||
|
||||
// Other options for implementations of the interface
|
||||
// can be stored in a context
|
||||
@@ -81,16 +89,24 @@ func Codec(c codec.Marshaler) Option {
|
||||
|
||||
// ErrorHandler will catch all broker errors that cant be handled
|
||||
// in normal way, for example Codec errors
|
||||
func HandleError(h ErrorHandler) SubscribeOption {
|
||||
func ErrorHandler(h Handler) Option {
|
||||
return func(o *Options) {
|
||||
o.ErrorHandler = h
|
||||
}
|
||||
}
|
||||
|
||||
// SubscribeErrorHandler will catch all broker errors that cant be handled
|
||||
// in normal way, for example Codec errors
|
||||
func SubscribeErrorHandler(h Handler) SubscribeOption {
|
||||
return func(o *SubscribeOptions) {
|
||||
o.ErrorHandler = h
|
||||
}
|
||||
}
|
||||
|
||||
// Queue sets the name of the queue to share messages on
|
||||
func Queue(name string) SubscribeOption {
|
||||
// SubscribeGroup sets the name of the queue to share messages on
|
||||
func SubscribeGroup(name string) SubscribeOption {
|
||||
return func(o *SubscribeOptions) {
|
||||
o.Queue = name
|
||||
o.Group = name
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user