code: func (s *subscriber) recv() {
  	// Close the connection once the subscriber stops receiving.
  	defer s.conn.Close()
  
  	for {
  		switch x := s.conn.Receive().(type) {
  		case redis.Message:
  			var m broker.Message
  
  			// Handle error? Only a log would be necessary since this type
  			// of issue cannot be fixed.
  			if err := s.codec.Unmarshal(x.Data, &m); err != nil {
  				break
  			}
  
  			p := publication{
  				topic:   x.Channel,
  				message: &m,
  			}
  
  			// Handle error? Retry?
  			if err := s.handle(&p); err != nil {
  				break
  			}
  
  			// Added for posterity, however Ack is a no-op.
  			if s.opts.AutoAck {
  				if err := p.Ack(); err != nil {
  					break
  				}
  			}
  
  		case redis.Subscription:
  			if x.Count == 0 {
  				return
  			}
  
  		case error:
  			return
  		}
  	}
  }
docs: // recv loops to receive new messages from Redis and handle them
  // as publications.