func WithSubscriberPoolSize(value uint) Option {
	return func(opts *options) {
		opts.subscriberPoolSize = value
	}
}
WithSubscriberPoolSize sets the size of the message queue subscriber pool.
package observer
import (
	"context"
	"encoding/json"
	"fmt"
	"net/url"
	"strings"
	"time"
	"github.com/ThreeDotsLabs/watermill/message"
	"github.com/hyperledger/aries-framework-go/pkg/doc/verifiable"
	"github.com/piprate/json-gold/ld"
	"github.com/trustbloc/edge-core/pkg/log"
	"github.com/trustbloc/sidetree-core-go/pkg/api/operation"
	"github.com/trustbloc/sidetree-core-go/pkg/api/protocol"
	txnapi "github.com/trustbloc/sidetree-core-go/pkg/api/txn"
	"github.com/trustbloc/orb/pkg/activitypub/vocab"
	"github.com/trustbloc/orb/pkg/anchor/anchorevent"
	"github.com/trustbloc/orb/pkg/anchor/graph"
	anchorinfo "github.com/trustbloc/orb/pkg/anchor/info"
	"github.com/trustbloc/orb/pkg/anchor/subject"
	"github.com/trustbloc/orb/pkg/anchor/util"
	discoveryrest "github.com/trustbloc/orb/pkg/discovery/endpoint/restapi"
	"github.com/trustbloc/orb/pkg/errors"
	"github.com/trustbloc/orb/pkg/hashlink"
	"github.com/trustbloc/orb/pkg/pubsub/spi"
)
var logger = log.New("orb-observer")
const defaultSubscriberPoolSize = 5
type AnchorGraph interface {
	Read(hl string) (*vocab.AnchorEventType, error)
	GetDidAnchors(cid, suffix string) ([]graph.Anchor, error)
}
type OperationStore interface {
	Put(ops []*operation.AnchoredOperation) error
}
type OperationFilter interface {
	Filter(uniqueSuffix string, ops []*operation.AnchoredOperation) ([]*operation.AnchoredOperation, error)
}
type didAnchors interface {
	PutBulk(dids []string, cid string) error
}
type Publisher interface {
	PublishAnchor(anchor *anchorinfo.AnchorInfo) error
	PublishDID(did string) error
}
type pubSub interface {
	Subscribe(ctx context.Context, topic string) (<-chan *message.Message, error)
	SubscribeWithOpts(ctx context.Context, topic string, opts ...spi.Option) (<-chan *message.Message, error)
	Publish(topic string, messages ...*message.Message) error
	Close() error
}
type metricsProvider interface {
	ProcessAnchorTime(value time.Duration)
	ProcessDIDTime(value time.Duration)
}
type Outbox interface {
	Post(activity *vocab.ActivityType) (*url.URL, error)
}
type resourceResolver interface {
	ResolveHostMetaLink(uri, linkType string) (string, error)
}
type casResolver interface {
	Resolve(webCASURL *url.URL, hl string, data []byte) ([]byte, string, error)
}
type documentLoader interface {
	LoadDocument(u string) (*ld.RemoteDocument, error)
}
type anchorLinkStore interface {
	PutLinks(links []*url.URL) error
}
type outboxProvider func() Outbox
type options struct {
	discoveryDomain    string
	subscriberPoolSize uint
}
type Option func(opts *options)
func WithDiscoveryDomain(domain string) Option {
	return func(opts *options) {
		opts.discoveryDomain = domain
	}
}