Method Loop

client.go:113–200

Loop should be run as condition for `for` with receiving from (*Client).Errors() It will manage AMQP connection, run queue and exchange declarations, consumers. Will start to return false once (*Client).Close() called.

()

// It will manage AMQP connection, run queue and exchange declarations, consumers.
// Will start to return false once (*Client).Close() called.
func (c *Client) Loop() bool {
	var (
		err error
	)

	if atomic.LoadInt32(&c.run) == noRun {
		return false
	}

	conn, _ := c.conn.Load().(*amqp.Connection)

	if conn != nil {
		return true
	}

	if c.bo != nil {
		time.Sleep(c.bo.Backoff(int(c.attempt)))
		atomic.AddInt32(&c.attempt, 1)
	}

	// set default Heartbeat to 10 seconds like in original amqp.Dial
	if c.config.Heartbeat == 0 {
		c.config.Heartbeat = 10 * time.Second
	}

	conn, err = amqp.DialConfig(c.addr, c.config)

	if c.reportErr(err) {
		return true
	}
	c.conn.Store(conn)

	atomic.StoreInt32(&c.attempt, 0)

	// guard conn
	go func() {
		chanErr := make(chan *amqp.Error)
		chanBlocking := make(chan amqp.Blocking)
		conn.NotifyClose(chanErr)
		conn.NotifyBlocked(chanBlocking)

		// loop for blocking/deblocking
		for {
			select {
			case err1 := <-chanErr:
				c.reportErr(err1)

				if conn1 := c.conn.Load().(*amqp.Connection); conn1 != nil {
					c.conn.Store((*amqp.Connection)(nil))
					conn1.Close()
				}
				// return from routine to launch reconnect process
				return
			case blocking := <-chanBlocking:
				select {
				case c.blocking <- blocking:
				default:
				}