// SubscribeWithOpts subscribes to a topic using the given options, and returns the Go channel over which messages
// are sent. The returned channel will be closed when Close() is called on this struct.
SubscribeWithOpts subscribes to a topic using the given options, and returns the Go channel over which messages
are sent. The returned channel will be closed when Close() is called on this struct.
// SubscribeWithOpts subscribes to a topic using the given options and returns the Go channel over which messages // are sent . The returned channel will be closed when Close () is called on this struct .
func (p *PubSub) SubscribeWithOpts(ctx context.Context, topic string,
	opts ...spi.Option) (<-chan *message.Message, error) {
	if p.State() != lifecycle.StateStarted {
		return nil, lifecycle.ErrNotStarted
	}
	options := &spi.Options{}
	for _, opt := range opts {
		opt(options)
	}
	if options.PoolSize <= 1 {
		logger.Debugf("Subscribing to topic [%s]", topic)
		return p.subscriber.Subscribe(ctx, topic)
	}
	logger.Debugf("Creating subscriber pool for topic [%s], Size [%d]", topic, options.PoolSize)
	pool, err := newPooledSubscriber(ctx, options.PoolSize, p.subscriber, topic)
	if err != nil {
		return nil, fmt.Errorf("subscriber pool: %w", err)
	}
	p.mutex.Lock()
	p.pools = append(p.pools, pool)
	p.mutex.Unlock()
	pool.start()
	return pool.msgChan, nil
}
func ( p * PubSub ) SubscribeWithOpts ( ctx context . Context , topic string , opts ... spi . Option ) ( <- chan * message . Message , error ) { if p . State ( ) != lifecycle . StateStarted { return nil , lifecycle . ErrNotStarted } options := & spi . Options { } for _ , opt := range opts { opt ( options ) } if options . PoolSize <= 1 { logger . Debugf ( " " , topic ) return p . subscriber . Subscribe ( ctx , topic ) } logger . Debugf ( " " , topic , options . PoolSize ) pool , err := newPooledSubscriber ( ctx , options . PoolSize , p . subscriber , topic ) if err != nil { return nil , fmt . Errorf ( " " , err ) } p . mutex . Lock ( ) p . pools = append ( p . pools , pool ) p . mutex . Unlock ( ) pool . start ( ) return pool . msgChan , nil }
SubscribeWithOpts subscribes to a topic using the given options, and returns the Go channel over which messages
are sent.
SubscribeWithOpts subscribes to a topic using the given options and returns the Go channel over which messages are sent .