func (p *PubSub) Publish(topic string, messages ...*message.Message) error {
	if p.State() != lifecycle.StateStarted {
		return lifecycle.ErrNotStarted
	}
	p.publishChan <- &entry{
		topic:    topic,
		messages: messages,
	}
	return nil
}
Publish publishes the given messages to the given topic. This function returns
immediately after sending the messages to the Go channel(s), although it will
block if the concurrency limit (defined by Config.Concurrency) has been reached.
package mempubsub
import (
	"context"
	"sync"
	"time"
	"github.com/ThreeDotsLabs/watermill/message"
	"github.com/trustbloc/edge-core/pkg/log"
	"github.com/trustbloc/orb/pkg/lifecycle"
	"github.com/trustbloc/orb/pkg/pubsub/spi"
)
var logger = log.New("pubsub")
const (
	defaultTimeout     = 10 * time.Second
	defaultConcurrency = 20
	defaultBufferSize  = 20
)
type Config struct {
	Timeout time.Duration
	Concurrency int
	BufferSize int
}
func DefaultConfig() Config {
	return Config{
		Timeout:     defaultTimeout,
		Concurrency: defaultConcurrency,
		BufferSize:  defaultBufferSize,
	}
}
type PubSub struct {
	*lifecycle.Lifecycle
	Config
	msgChansByTopic map[string][]chan *message.Message
	mutex           sync.RWMutex
	publishChan     chan *entry
	ackChan         chan *message.Message
	doneChan        chan struct{}
}
type entry struct {
	topic    string
	messages []*message.Message
}
func New(cfg Config) *PubSub {
	m := &PubSub{
		Config:          cfg,
		msgChansByTopic: make(map[string][]chan *message.Message),
		publishChan:     make(chan *entry, cfg.BufferSize),
		ackChan:         make(chan *message.Message, cfg.Concurrency),
		doneChan:        make(chan struct{}),
	}
	m.Lifecycle = lifecycle.New("httpsubscriber", lifecycle.WithStop(m.stop))
	go m.processMessages()
	go m.processAcks()
	m.Start()
	return m
}
func (p *PubSub) Close() error {
	p.Stop()
	return nil
}
func (p *PubSub) stop() {
	logger.Infof("Stopping publisher/subscriber...")
	p.doneChan <- struct{}{}
	logger.Debugf("... waiting for publisher to stop...")
	<-p.doneChan
	logger.Debugf("... closing subscriber channels...")
	p.mutex.Lock()
	defer p.mutex.Unlock()
	for _, msgChans := range p.msgChansByTopic {
		for _, msgChan := range msgChans {
			close(msgChan)
		}
	}
	p.msgChansByTopic = nil
	close(p.ackChan)
	logger.Infof("... publisher/subscriber stopped.")
}
func (p *PubSub) Subscribe(ctx context.Context, topic string) (<-chan *message.Message, error) {
	return p.SubscribeWithOpts(ctx, topic)
}
func (p *PubSub) SubscribeWithOpts(_ context.Context, topic string, _ ...spi.Option) (<-chan *message.Message, error) {
	if p.State() != lifecycle.StateStarted {
		return nil, lifecycle.ErrNotStarted
	}
	logger.Debugf("Subscribing to topic [%s]", topic)
	p.mutex.Lock()
	defer p.mutex.Unlock()
	msgChan := make(chan *message.Message, p.BufferSize)
	p.msgChansByTopic[topic] = append(p.msgChansByTopic[topic], msgChan)
	return msgChan, nil
}