code: func (b *Broker) Publish(ctx context.Context, signature *tasks.Signature) error {
  	// Adjust routing key (this decides which queue the message will be published to)
  	b.AdjustRoutingKey(signature)
  
  	msg, err := json.Marshal(signature)
  	if err != nil {
  		return fmt.Errorf("JSON marshal error: %s", err)
  	}
  
  	// Check the ETA signature field, if it is set and it is in the future,
  	// delay the task
  	if signature.ETA != nil {
  		now := time.Now().UTC()
  
  		if signature.ETA.After(now) {
  			delayMs := int64(signature.ETA.Sub(now) / time.Millisecond)
  
  			return b.delay(signature, delayMs)
  		}
  	}
  
  	connection, err := b.GetOrOpenConnection(signature.RoutingKey,
  		b.GetConfig().AMQP.BindingKey, // queue binding key
  		nil,                           // exchange declare args
  		nil,                           // queue declare args
  		amqp.Table(b.GetConfig().AMQP.QueueBindingArgs), // queue binding args
  	)
  	if err != nil {
  		return err
  	}
  
  	channel := connection.channel
  	confirmsChan := connection.confirmation
  
  	if err := channel.Publish(
  		b.GetConfig().AMQP.Exchange, // exchange name
  		signature.RoutingKey,        // routing key
  		false,                       // mandatory
  		false,                       // immediate
  		amqp.Publishing{
  			Headers:      amqp.Table(signature.Headers),
  			ContentType:  "application/json",
  			Body:         msg,
  			DeliveryMode: amqp.Persistent,
  		},
  	); err != nil {
  		return err
  	}
  
  	confirmed := <-confirmsChan
  
  	if confirmed.Ack {
  		return nil
  	}
  
  	return fmt.Errorf("Failed delivery of delivery tag: %v", confirmed.DeliveryTag)
  }
docs: // Publish places a new message on the default queue