code: func (b *pubsubBroker) Subscribe(topic string, h broker.Handler, opts ...broker.SubscribeOption) (broker.Subscriber, error) {
  	if b.err != nil {
  		return nil, b.err
  	}
  	options := broker.SubscribeOptions{
  		AutoAck: false,
  		Context: b.options.Context,
  	}
  	for _, o := range opts {
  		o(&options)
  	}
  	if options.Queue == "" {
  		return nil, errors.New("gocloud.Subscribe: need Queue option")
  	}
  	t := b.topic(topic)
  	b.mu.Lock()
  	defer b.mu.Unlock()
  	s := b.subs[options.Queue]
  	if s == nil {
  		// TODO(jba): how to configure ack deadline?
  		s = b.openSub(t, options.Queue)
  		b.subs[options.Queue] = s
  	}
  	// TODO(jba): how can we verify that an existing subscription's topic matches?
  	ctx, cancel := context.WithCancel(context.Background())
  	sub := &subscriber{
  		options: options,
  		topic:   topic,
  		sub:     s,
  		cancel:  cancel,
  	}
  	go sub.run(ctx, h)
  	return sub, nil
  }
docs: // Subscribe opens a subscription to the given topic and begins receiving messages and passing
  // them to handler in a separate goroutine.
  //
  // The Queue SubscribeOption is required. Subscribe never creates new subscriptions on the backend (except
  // for mempubsub, but you still need a queue name).