merge master!

Author: asim

README.md

@@ -45,10 +45,10 @@ in the plugins repo. State and persistence becomes a core requirement beyond pro
 
 ## Getting Started
 
-To make use of Go Micro import it 
+To make use of Go Micro 
 
 ```golang
-import "go-micro.dev/v5"
+go get "go-micro.dev/v5"
 ```
 
 Create a service and register a handler
@@ -105,3 +105,7 @@ curl -XPOST \
      -d '{"name": "alice"}' \
       http://localhost:8080
 ```
+
+## Adopters
+
+- [Sourse](https://sourse.eu) - Work in the field of earth observation, including embedded Kubernetes running onboard aircraft, and we’ve built a mission management SaaS platform using Go Micro.

broker/broker.go

@@ -41,7 +41,7 @@ type Subscriber interface {
 
 var (
 	// DefaultBroker is the default Broker.
-	DefaultBroker = NewBroker()
+	DefaultBroker = NewMemoryBroker()
 )
 
 func Init(opts ...Option) error {

broker/http.go renamed to broker/http/http.go

@@ -1,5 +1,5 @@
-// Package broker provides a http based message broker
-package broker
+// Package http provides a http based message broker
+package http
 
 import (
 	"bytes"
@@ -16,6 +16,7 @@ import (
 	"time"
 
 	"github.com/google/uuid"
+	"go-micro.dev/v5/broker"
 	"go-micro.dev/v5/codec/json"
 	merr "go-micro.dev/v5/errors"
 	"go-micro.dev/v5/registry"
@@ -29,7 +30,7 @@ import (
 
 // HTTP Broker is a point to point async broker.
 type httpBroker struct {
-	opts Options
+	opts broker.Options
 
 	r registry.Registry
 
@@ -51,8 +52,8 @@ type httpBroker struct {
 }
 
 type httpSubscriber struct {
-	opts  SubscribeOptions
-	fn    Handler
+	opts  broker.SubscribeOptions
+	fn    broker.Handler
 	svc   *registry.Service
 	hb    *httpBroker
 	id    string
@@ -61,7 +62,7 @@ type httpSubscriber struct {
 
 type httpEvent struct {
 	err error
-	m   *Message
+	m   *broker.Message
 	t   string
 }
 
@@ -108,8 +109,8 @@ func newTransport(config *tls.Config) *http.Transport {
 	return t
 }
 
-func newHttpBroker(opts ...Option) Broker {
-	options := *NewOptions(opts...)
+func newHttpBroker(opts ...broker.Option) broker.Broker {
+	options := *broker.NewOptions(opts...)
 
 	options.Registry = registry.DefaultRegistry
 	options.Codec = json.Marshaler{}
@@ -161,15 +162,15 @@ func (h *httpEvent) Error() error {
 	return h.err
 }
 
-func (h *httpEvent) Message() *Message {
+func (h *httpEvent) Message() *broker.Message {
 	return h.m
 }
 
 func (h *httpEvent) Topic() string {
 	return h.t
 }
 
-func (h *httpSubscriber) Options() SubscribeOptions {
+func (h *httpSubscriber) Options() broker.SubscribeOptions {
 	return h.opts
 }
 
@@ -308,7 +309,7 @@ func (h *httpBroker) ServeHTTP(w http.ResponseWriter, req *http.Request) {
 		return
 	}
 
-	var m *Message
+	var m *broker.Message
 	if err = h.opts.Codec.Unmarshal(b, &m); err != nil {
 		errr := merr.InternalServerError("go.micro.broker", "Error parsing request body: %v", err)
 		w.WriteHeader(500)
@@ -330,7 +331,7 @@ func (h *httpBroker) ServeHTTP(w http.ResponseWriter, req *http.Request) {
 	id := req.Form.Get("id")
 
 	//nolint:prealloc
-	var subs []Handler
+	var subs []broker.Handler
 
 	h.RLock()
 	for _, subscriber := range h.subscribers[topic] {
@@ -458,7 +459,7 @@ func (h *httpBroker) Disconnect() error {
 	return err
 }
 
-func (h *httpBroker) Init(opts ...Option) error {
+func (h *httpBroker) Init(opts ...broker.Option) error {
 	h.RLock()
 	if h.running {
 		h.RUnlock()
@@ -505,13 +506,13 @@ func (h *httpBroker) Init(opts ...Option) error {
 	return nil
 }
 
-func (h *httpBroker) Options() Options {
+func (h *httpBroker) Options() broker.Options {
 	return h.opts
 }
 
-func (h *httpBroker) Publish(topic string, msg *Message, opts ...PublishOption) error {
+func (h *httpBroker) Publish(topic string, msg *broker.Message, opts ...broker.PublishOption) error {
 	// create the message first
-	m := &Message{
+	m := &broker.Message{
 		Header: make(map[string]string),
 		Body:   msg.Body,
 	}
@@ -637,10 +638,10 @@ func (h *httpBroker) Publish(topic string, msg *Message, opts ...PublishOption)
 	return nil
 }
 
-func (h *httpBroker) Subscribe(topic string, handler Handler, opts ...SubscribeOption) (Subscriber, error) {
+func (h *httpBroker) Subscribe(topic string, handler broker.Handler, opts ...broker.SubscribeOption) (broker.Subscriber, error) {
 	var err error
 	var host, port string
-	options := NewSubscribeOptions(opts...)
+	options := broker.NewSubscribeOptions(opts...)
 
 	// parse address for host, port
 	host, port, err = net.SplitHostPort(h.Address())
@@ -705,7 +706,7 @@ func (h *httpBroker) String() string {
 	return "http"
 }
 
-// NewBroker returns a new http broker.
-func NewBroker(opts ...Option) Broker {
+// NewHttpBroker returns a new http broker.
+func NewHttpBroker(opts ...broker.Option) broker.Broker {
 	return newHttpBroker(opts...)
 }

broker/http_test.go renamed to broker/http/http_test.go

@@ -1,4 +1,4 @@
-package broker_test
+package http_test
 
 import (
 	"sync"
@@ -7,6 +7,7 @@ import (
 
 	"github.com/google/uuid"
 	"go-micro.dev/v5/broker"
+	"go-micro.dev/v5/broker/http"
 	"go-micro.dev/v5/registry"
 )
 
@@ -60,7 +61,7 @@ func sub(b *testing.B, c int) {
 	b.StopTimer()
 	m := newTestRegistry()
 
-	brker := broker.NewBroker(broker.Registry(m))
+	brker := http.NewHttpBroker(broker.Registry(m))
 	topic := uuid.New().String()
 
 	if err := brker.Init(); err != nil {
@@ -121,7 +122,7 @@ func sub(b *testing.B, c int) {
 func pub(b *testing.B, c int) {
 	b.StopTimer()
 	m := newTestRegistry()
-	brk := broker.NewBroker(broker.Registry(m))
+	brk := http.NewHttpBroker(broker.Registry(m))
 	topic := uuid.New().String()
 
 	if err := brk.Init(); err != nil {
@@ -190,7 +191,7 @@ func pub(b *testing.B, c int) {
 
 func TestBroker(t *testing.T) {
 	m := newTestRegistry()
-	b := broker.NewBroker(broker.Registry(m))
+	b := http.NewHttpBroker(broker.Registry(m))
 
 	if err := b.Init(); err != nil {
 		t.Fatalf("Unexpected init error: %v", err)
@@ -239,7 +240,7 @@ func TestBroker(t *testing.T) {
 
 func TestConcurrentSubBroker(t *testing.T) {
 	m := newTestRegistry()
-	b := broker.NewBroker(broker.Registry(m))
+	b := http.NewHttpBroker(broker.Registry(m))
 
 	if err := b.Init(); err != nil {
 		t.Fatalf("Unexpected init error: %v", err)
@@ -298,7 +299,7 @@ func TestConcurrentSubBroker(t *testing.T) {
 
 func TestConcurrentPubBroker(t *testing.T) {
 	m := newTestRegistry()
-	b := broker.NewBroker(broker.Registry(m))
+	b := http.NewHttpBroker(broker.Registry(m))
 
 	if err := b.Init(); err != nil {
 		t.Fatalf("Unexpected init error: %v", err)

broker/nats/context.go

@@ -0,0 +1,17 @@
+package nats
+
+import (
+	"context"
+
+	"go-micro.dev/v5/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)
+	}
+}