func (q *Queue) Len() uint {
	if q.State() != lifecycle.StateStarted {
		return 0
	}
	q.mutex.RLock()
	defer q.mutex.RUnlock()
	return uint(len(q.pending))
}
Len returns the length of the pending queue.
package opqueue
import (
	"context"
	"encoding/json"
	"fmt"
	"sync"
	"time"
	"github.com/ThreeDotsLabs/watermill"
	"github.com/ThreeDotsLabs/watermill/message"
	"github.com/trustbloc/edge-core/pkg/log"
	"github.com/trustbloc/sidetree-core-go/pkg/api/operation"
	"github.com/trustbloc/orb/pkg/lifecycle"
	"github.com/trustbloc/orb/pkg/pubsub/spi"
)
var logger = log.New("sidetree_context")
const topic = "opqueue"
type pubSub interface {
	SubscribeWithOpts(ctx context.Context, topic string, opts ...spi.Option) (<-chan *message.Message, error)
	Publish(topic string, messages ...*message.Message) error
	Close() error
}
type operationMessage struct {
	msg       *message.Message
	op        *operation.QueuedOperationAtTime
	timeAdded time.Time
}
type metricsProvider interface {
	AddOperationTime(value time.Duration)
	BatchCutTime(value time.Duration)
	BatchRollbackTime(value time.Duration)
	BatchAckTime(value time.Duration)
	BatchNackTime(value time.Duration)
	BatchSize(value float64)
}
type Config struct {
	PoolSize uint
}
type Queue struct {
	*lifecycle.Lifecycle
	pubSub        pubSub
	msgChan       <-chan *message.Message
	mutex         sync.RWMutex
	pending       []*operationMessage
	jsonMarshal   func(interface{}) ([]byte, error)
	jsonUnmarshal func(data []byte, v interface{}) error
	metrics       metricsProvider
}
func New(cfg Config, pubSub pubSub, metrics metricsProvider) (*Queue, error) {
	msgChan, err := pubSub.SubscribeWithOpts(context.Background(), topic, spi.WithPool(cfg.PoolSize))
	if err != nil {
		return nil, fmt.Errorf("subscribe to topic [%s]: %w", topic, err)
	}
	q := &Queue{
		pubSub:        pubSub,
		msgChan:       msgChan,
		jsonMarshal:   json.Marshal,
		jsonUnmarshal: json.Unmarshal,
		metrics:       metrics,
	}
	q.Lifecycle = lifecycle.New("operation-queue",
		lifecycle.WithStart(q.start),
		lifecycle.WithStop(q.stop),
	)
	q.Start()
	return q, nil
}
func (q *Queue) Add(op *operation.QueuedOperation, protocolVersion uint64) (uint, error) {
	if q.State() != lifecycle.StateStarted {
		return 0, lifecycle.ErrNotStarted
	}
	startTime := time.Now()
	defer func() {
		q.metrics.AddOperationTime(time.Since(startTime))
	}()
	b, err := q.jsonMarshal(
		&operation.QueuedOperationAtTime{
			QueuedOperation: *op,
			ProtocolVersion: protocolVersion,
		},
	)
	if err != nil {
		return 0, fmt.Errorf("marshall queued operation: %w", err)
	}
	msg := message.NewMessage(watermill.NewUUID(), b)
	logger.Debugf("Publishing operation message to topic [%s] - Msg [%s], DID [%s]", topic, msg.UUID, op.UniqueSuffix)
	err = q.pubSub.Publish(topic, msg)
	if err != nil {
		return 0, fmt.Errorf("publish queued operation: %w", err)
	}
	q.mutex.RLock()
	defer q.mutex.RUnlock()
	return uint(len(q.pending)), nil
}
func (q *Queue) Peek(num uint) (operation.QueuedOperationsAtTime, error) {
	if q.State() != lifecycle.StateStarted {
		return nil, lifecycle.ErrNotStarted
	}
	q.mutex.RLock()
	defer q.mutex.RUnlock()
	n := int(num)
	if len(q.pending) < n {
		n = len(q.pending)
	}
	return asQueuedOperations(q.pending[0:n]), nil
}
func (q *Queue) Remove(num uint) (ops operation.QueuedOperationsAtTime, ack func() uint, nack func(), err error) {
	if q.State() != lifecycle.StateStarted {
		return nil, nil, nil, lifecycle.ErrNotStarted
	}
	startTime := time.Now()
	q.mutex.Lock()
	defer q.mutex.Unlock()
	n := int(num)
	if len(q.pending) < n {
		n = len(q.pending)
	}
	if n == 0 {
		return nil,
			func() uint { return 0 },
			func() {}, nil
	}
	items := q.pending[0:n]
	q.pending = q.pending[n:]
	return asQueuedOperations(items), q.newAckFunc(items, startTime), q.newNackFunc(items, startTime), nil
}