code: func (r *Rabbus) Run(ctx context.Context) error {
  	notifyClose := r.NotifyClose(make(chan *amqp.Error))
  
  	for {
  		select {
  		case m, ok := <-r.emit:
  			if !ok {
  				return errors.New("unexpected close of emit channel")
  			}
  
  			r.produce(m)
  
  		case err := <-notifyClose:
  			if err == nil {
  				// "… on a graceful close, no error will be sent."
  				return nil
  			}
  
  			r.handleAmqpClose(err)
  
  			// We have reconnected, so we need a new NotifyClose again.
  			notifyClose = r.NotifyClose(make(chan *amqp.Error))
  
  		case <-ctx.Done():
  			return ctx.Err()
  		}
  	}
  }
docs: // Run starts rabbus channels for emitting and listening for amqp connection close
  // returns ctx error in case of any.