func (m *MockPubSub) Close() error {
	if m.Err != nil {
		return m.Err
	}
	m.Stop()
	return nil
}
Close closes the subscriber channels.
package mocks
import (
	"context"
	"sync"
	"time"
	"github.com/ThreeDotsLabs/watermill/message"
	"github.com/trustbloc/orb/pkg/lifecycle"
	"github.com/trustbloc/orb/pkg/pubsub/spi"
)
const (
	maxBufferSize = 5
	timeout       = 100 * time.Millisecond
)
type MockPubSub struct {
	*lifecycle.Lifecycle
	Err               error
	MsgChan           map[string]chan *message.Message
	mutex             sync.RWMutex
	Timeout           time.Duration
	undeliverableChan chan *message.Message
	done              chan struct{}
}
func NewPubSub() *MockPubSub {
	m := &MockPubSub{
		MsgChan:           make(map[string]chan *message.Message),
		Timeout:           timeout,
		undeliverableChan: make(chan *message.Message, maxBufferSize),
		done:              make(chan struct{}),
	}
	m.Lifecycle = lifecycle.New("mock_pubsub", lifecycle.WithStop(m.stop))
	go m.handleUndeliverable()
	m.Start()
	return m
}
func (m *MockPubSub) WithError(err error) *MockPubSub {
	m.Err = err
	return m
}
func (m *MockPubSub) Subscribe(_ context.Context, topic string) (<-chan *message.Message, error) {
	if m.Err != nil {
		return nil, m.Err
	}
	m.mutex.Lock()
	defer m.mutex.Unlock()
	if m.State() != lifecycle.StateStarted {
		return nil, lifecycle.ErrNotStarted
	}
	msgChan := make(chan *message.Message, maxBufferSize)
	m.MsgChan[topic] = msgChan
	return msgChan, nil
}
func (m *MockPubSub) SubscribeWithOpts(ctx context.Context,
	topic string, _ ...spi.Option) (<-chan *message.Message, error) {
	return m.Subscribe(ctx, topic)
}
func (m *MockPubSub) Publish(topic string, messages ...*message.Message) error {
	if m.Err != nil {
		return m.Err
	}
	m.mutex.RLock()
	defer m.mutex.RUnlock()
	if m.State() != lifecycle.StateStarted {
		return lifecycle.ErrNotStarted
	}
	msgChan := m.MsgChan[topic]
	for _, msg := range messages {
		msg = msg.Copy()
		msgChan <- msg
		go m.check(msg)
	}
	return nil
}