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
}
Remove removes (up to) the given number of items from the head of the queue.
Returns the actual number of items that were removed and the new length of the queue.
Each removed message is acknowledged, indicating that it was successfully processed. If the
server goes down with messages still in the queue then, if using a durable message queue, the messages
will be delivered to another server instance which is processing from the same 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
}