// NewPubSub returns a new publisher/subscriber.
NewPubSub returns a new publisher/subscriber.
// NewPubSub returns a new publisher / subscriber .
func NewPubSub(pubSub pubSub, anchorProcessor anchorProcessor, didProcessor didProcessor,
	poolSize int) (*PubSub, error) {
	h := &PubSub{
		publisher:      pubSub,
		processAnchors: anchorProcessor,
		processDID:     didProcessor,
		jsonUnmarshal:  json.Unmarshal,
		jsonMarshal:    json.Marshal,
	}
	h.Lifecycle = lifecycle.New("observer-pubsub",
		lifecycle.WithStart(h.start),
	)
	logger.Infof("Subscribing to topic [%s]", anchorTopic)
	anchorCredChan, err := pubSub.SubscribeWithOpts(context.Background(), anchorTopic, spi.WithPool(poolSize))
	if err != nil {
		return nil, fmt.Errorf("subscribe to topic [%s]: %w", anchorTopic, err)
	}
	h.anchorCredChan = anchorCredChan
	logger.Infof("Subscribing to topic [%s]", didTopic)
	didChan, err := pubSub.SubscribeWithOpts(context.Background(), didTopic, spi.WithPool(poolSize))
	if err != nil {
		return nil, fmt.Errorf("subscribe to topic [%s]: %w", didTopic, err)
	}
	h.didChan = didChan
	return h, nil
}
func NewPubSub ( pubSub pubSub , anchorProcessor anchorProcessor , didProcessor didProcessor , poolSize int ) ( * PubSub , error ) { h := & PubSub { publisher : pubSub , processAnchors : anchorProcessor , processDID : didProcessor , jsonUnmarshal : json . Unmarshal , jsonMarshal : json . Marshal , } h . Lifecycle = lifecycle . New ( " " , lifecycle . WithStart ( h . start ) , ) logger . Infof ( " " , anchorTopic ) anchorCredChan , err := pubSub . SubscribeWithOpts ( context . Background ( ) , anchorTopic , spi . WithPool ( poolSize ) ) if err != nil { return nil , fmt . Errorf ( " " , anchorTopic , err ) } h . anchorCredChan = anchorCredChan logger . Infof ( " " , didTopic ) didChan , err := pubSub . SubscribeWithOpts ( context . Background ( ) , didTopic , spi . WithPool ( poolSize ) ) if err != nil { return nil , fmt . Errorf ( " " , didTopic , err ) } h . didChan = didChan return h , nil }
NewPubSub returns a new publisher/subscriber.
NewPubSub returns a new publisher / subscriber .