code: func (b *redisBroker) Subscribe(topic string, handler broker.Handler, opts ...broker.SubscribeOption) (broker.Subscriber, error) {
  	var options broker.SubscribeOptions
  	for _, o := range opts {
  		o(&options)
  	}
  
  	s := subscriber{
  		codec:  b.opts.Codec,
  		conn:   &redis.PubSubConn{Conn: b.pool.Get()},
  		topic:  topic,
  		handle: handler,
  		opts:   options,
  	}
  
  	// Run the receiver routine.
  	go s.recv()
  
  	if err := s.conn.Subscribe(s.topic); err != nil {
  		return nil, err
  	}
  
  	return &s, nil
  }
docs: // Subscribe returns a subscriber for the topic and handler.