micro/broker/mock/mock.go

180 lines
2.8 KiB
Go
Raw Normal View History

2016-03-18 01:46:20 +03:00
package mock
import (
"errors"
"sync"
"github.com/google/uuid"
2016-03-18 01:46:20 +03:00
"github.com/micro/go-micro/broker"
)
2016-03-18 01:52:49 +03:00
type mockBroker struct {
2016-03-18 01:46:20 +03:00
opts broker.Options
sync.RWMutex
connected bool
2016-03-18 01:52:49 +03:00
Subscribers map[string][]*mockSubscriber
2016-03-18 01:46:20 +03:00
}
2016-03-18 01:52:49 +03:00
type mockPublication struct {
2016-03-18 01:46:20 +03:00
topic string
message *broker.Message
}
2016-03-18 01:52:49 +03:00
type mockSubscriber struct {
2016-03-18 01:46:20 +03:00
id string
topic string
exit chan bool
handler broker.Handler
opts broker.SubscribeOptions
}
2016-03-18 01:52:49 +03:00
func (m *mockBroker) Options() broker.Options {
2016-03-18 01:46:20 +03:00
return m.opts
}
2016-03-18 01:52:49 +03:00
func (m *mockBroker) Address() string {
2016-03-18 01:46:20 +03:00
return ""
}
2016-03-18 01:52:49 +03:00
func (m *mockBroker) Connect() error {
2016-03-18 01:46:20 +03:00
m.Lock()
defer m.Unlock()
if m.connected {
return nil
}
m.connected = true
return nil
}
2016-03-18 01:52:49 +03:00
func (m *mockBroker) Disconnect() error {
2016-03-18 01:46:20 +03:00
m.Lock()
defer m.Unlock()
if !m.connected {
return nil
}
m.connected = false
return nil
}
2016-03-18 01:52:49 +03:00
func (m *mockBroker) Init(opts ...broker.Option) error {
2016-03-18 01:46:20 +03:00
for _, o := range opts {
o(&m.opts)
}
return nil
}
2016-03-18 01:52:49 +03:00
func (m *mockBroker) Publish(topic string, message *broker.Message, opts ...broker.PublishOption) error {
2016-03-18 01:46:20 +03:00
m.Lock()
defer m.Unlock()
if !m.connected {
return errors.New("not connected")
}
subs, ok := m.Subscribers[topic]
if !ok {
return nil
}
2016-03-18 01:52:49 +03:00
p := &mockPublication{
2016-03-18 01:46:20 +03:00
topic: topic,
message: message,
}
for _, sub := range subs {
if err := sub.handler(p); err != nil {
return err
}
}
return nil
}
2016-03-18 01:52:49 +03:00
func (m *mockBroker) Subscribe(topic string, handler broker.Handler, opts ...broker.SubscribeOption) (broker.Subscriber, error) {
2016-03-18 01:46:20 +03:00
m.Lock()
defer m.Unlock()
if !m.connected {
return nil, errors.New("not connected")
}
var options broker.SubscribeOptions
for _, o := range opts {
o(&options)
}
2016-03-18 01:52:49 +03:00
sub := &mockSubscriber{
2016-03-18 01:46:20 +03:00
exit: make(chan bool, 1),
id: uuid.New().String(),
2016-03-18 01:46:20 +03:00
topic: topic,
handler: handler,
opts: options,
}
m.Subscribers[topic] = append(m.Subscribers[topic], sub)
go func() {
<-sub.exit
m.Lock()
2016-03-18 01:52:49 +03:00
var newSubscribers []*mockSubscriber
2016-03-18 01:46:20 +03:00
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
}
2016-03-18 01:52:49 +03:00
func (m *mockBroker) String() string {
2016-03-18 01:46:20 +03:00
return "mock"
}
2016-03-18 01:52:49 +03:00
func (m *mockPublication) Topic() string {
2016-03-18 01:46:20 +03:00
return m.topic
}
2016-03-18 01:52:49 +03:00
func (m *mockPublication) Message() *broker.Message {
2016-03-18 01:46:20 +03:00
return m.message
}
2016-03-18 01:52:49 +03:00
func (m *mockPublication) Ack() error {
2016-03-18 01:46:20 +03:00
return nil
}
2016-03-18 01:52:49 +03:00
func (m *mockSubscriber) Options() broker.SubscribeOptions {
2016-03-18 01:46:20 +03:00
return m.opts
}
2016-03-18 01:52:49 +03:00
func (m *mockSubscriber) Topic() string {
2016-03-18 01:46:20 +03:00
return m.topic
}
2016-03-18 01:52:49 +03:00
func (m *mockSubscriber) Unsubscribe() error {
2016-03-18 01:46:20 +03:00
m.exit <- true
return nil
}
func NewBroker(opts ...broker.Option) broker.Broker {
var options broker.Options
for _, o := range opts {
o(&options)
}
2016-03-18 01:52:49 +03:00
return &mockBroker{
2016-03-18 01:46:20 +03:00
opts: options,
2016-03-18 01:52:49 +03:00
Subscribers: make(map[string][]*mockSubscriber),
2016-03-18 01:46:20 +03:00
}
}