move memory implementations to core micro repo

Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>

broker/memory.go

@@ -0,0 +1,247 @@
+package broker
+
+import (
+	"context"
+	"errors"
+	"math/rand"
+	"sync"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/unistack-org/micro/v3/logger"
+	maddr "github.com/unistack-org/micro/v3/util/addr"
+	mnet "github.com/unistack-org/micro/v3/util/net"
+)
+
+type memoryBroker struct {
+	opts Options
+
+	addr string
+	sync.RWMutex
+	connected   bool
+	Subscribers map[string][]*memorySubscriber
+}
+
+type memoryEvent struct {
+	opts    Options
+	topic   string
+	err     error
+	message interface{}
+}
+
+type memorySubscriber struct {
+	id      string
+	topic   string
+	exit    chan bool
+	handler Handler
+	opts    SubscribeOptions
+	ctx     context.Context
+}
+
+func (m *memoryBroker) Options() Options {
+	return m.opts
+}
+
+func (m *memoryBroker) Address() string {
+	return m.addr
+}
+
+func (m *memoryBroker) Connect(ctx context.Context) 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(ctx context.Context) error {
+	m.Lock()
+	defer m.Unlock()
+
+	if !m.connected {
+		return nil
+	}
+
+	m.connected = false
+
+	return nil
+}
+
+func (m *memoryBroker) Init(opts ...Option) error {
+	for _, o := range opts {
+		o(&m.opts)
+	}
+	return nil
+}
+
+func (m *memoryBroker) Publish(ctx context.Context, topic string, msg *Message, opts ...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
+	}
+
+	var v interface{}
+	if m.opts.Codec != nil {
+		buf, err := m.opts.Codec.Marshal(msg)
+		if err != nil {
+			return err
+		}
+		v = buf
+	} else {
+		v = msg
+	}
+
+	p := &memoryEvent{
+		topic:   topic,
+		message: v,
+		opts:    m.opts,
+	}
+
+	eh := m.opts.ErrorHandler
+
+	for _, sub := range subs {
+		if err := sub.handler(p); err != nil {
+			p.err = err
+			if sub.opts.ErrorHandler != nil {
+				eh = sub.opts.ErrorHandler
+			}
+			if eh != nil {
+				eh(p)
+			} else {
+				if m.opts.Logger.V(logger.ErrorLevel) {
+					m.opts.Logger.Error(m.opts.Context, err.Error())
+				}
+			}
+			continue
+		}
+	}
+
+	return nil
+}
+
+func (m *memoryBroker) Subscribe(ctx context.Context, topic string, handler Handler, opts ...SubscribeOption) (Subscriber, error) {
+	m.RLock()
+	if !m.connected {
+		m.RUnlock()
+		return nil, errors.New("not connected")
+	}
+	m.RUnlock()
+
+	options := NewSubscribeOptions(opts...)
+
+	id, err := uuid.NewRandom()
+	if err != nil {
+		return nil, err
+	}
+
+	sub := &memorySubscriber{
+		exit:    make(chan bool, 1),
+		id:      id.String(),
+		topic:   topic,
+		handler: handler,
+		opts:    options,
+		ctx:     ctx,
+	}
+
+	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 *memoryBroker) Name() string {
+	return m.opts.Name
+}
+
+func (m *memoryEvent) Topic() string {
+	return m.topic
+}
+
+func (m *memoryEvent) Message() *Message {
+	switch v := m.message.(type) {
+	case *Message:
+		return v
+	case []byte:
+		msg := &Message{}
+		if err := m.opts.Codec.Unmarshal(v, msg); err != nil {
+			if m.opts.Logger.V(logger.ErrorLevel) {
+				m.opts.Logger.Error(m.opts.Context, "[memory]: failed to unmarshal: %v", err)
+			}
+			return nil
+		}
+		return msg
+	}
+
+	return nil
+}
+
+func (m *memoryEvent) Ack() error {
+	return nil
+}
+
+func (m *memoryEvent) Error() error {
+	return m.err
+}
+
+func (m *memorySubscriber) Options() SubscribeOptions {
+	return m.opts
+}
+
+func (m *memorySubscriber) Topic() string {
+	return m.topic
+}
+
+func (m *memorySubscriber) Unsubscribe(ctx context.Context) error {
+	m.exit <- true
+	return nil
+}
+
+func NewBroker(opts ...Option) Broker {
+	rand.Seed(time.Now().UnixNano())
+
+	return &memoryBroker{
+		opts:        NewOptions(opts...),
+		Subscribers: make(map[string][]*memorySubscriber),
+	}
+}