code: func publish(sessions chan chan session, messages <-chan message) {
  	for session := range sessions {
  		var (
  			running bool
  			reading = messages
  			pending = make(chan message, 1)
  			confirm = make(chan amqp.Confirmation, 1)
  		)
  
  		pub := <-session
  
  		// publisher confirms for this channel/connection
  		if err := pub.Confirm(false); err != nil {
  			log.Printf("publisher confirms not supported")
  			close(confirm) // confirms not supported, simulate by always nacking
  		} else {
  			pub.NotifyPublish(confirm)
  		}
  
  		log.Printf("publishing...")
  
  	Publish:
  		for {
  			var body message
  			select {
  			case confirmed, ok := <-confirm:
  				if !ok {
  					break Publish
  				}
  				if !confirmed.Ack {
  					log.Printf("nack message %d, body: %q", confirmed.DeliveryTag, string(body))
  				}
  				reading = messages
  
  			case body = <-pending:
  				routingKey := "ignored for fanout exchanges, application dependent for other exchanges"
  				err := pub.Publish(exchange, routingKey, false, false, amqp.Publishing{
  					Body: body,
  				})
  				// Retry failed delivery on the next session
  				if err != nil {
  					pending <- body
  					pub.Close()
  					break Publish
  				}
  
  			case body, running = <-reading:
  				// all messages consumed
  				if !running {
  					return
  				}
  				// work on pending delivery until ack'd
  				pending <- body
  				reading = nil
  			}
  		}
  	}
  }
docs: // publish publishes messages to a reconnecting session to a fanout exchange.
  // It receives from the application specific source of messages.