code: func (b *Broker) consumeOne(delivery *pubsub.Message, taskProcessor iface.TaskProcessor) error {
  	if len(delivery.Data) == 0 {
  		delivery.Nack()
  		log.ERROR.Printf("received an empty message, the delivery was %v", delivery)
  		return errors.New("Received an empty message")
  	}
  
  	sig := new(tasks.Signature)
  	decoder := json.NewDecoder(bytes.NewBuffer(delivery.Data))
  	decoder.UseNumber()
  	if err := decoder.Decode(sig); err != nil {
  		delivery.Nack()
  		log.ERROR.Printf("unmarshal error. the delivery is %v", delivery)
  		return err
  	}
  
  	// If the task is not registered return an error
  	// and leave the message in the queue
  	if !b.IsTaskRegistered(sig.Name) {
  		delivery.Nack()
  		return fmt.Errorf("task %s is not registered", sig.Name)
  	}
  
  	err := taskProcessor.Process(sig)
  	if err != nil {
  		delivery.Nack()
  		return err
  	}
  
  	// Call Ack() after successfully consuming and processing the message
  	delivery.Ack()
  
  	return err
  }
docs: // consumeOne processes a single message using TaskProcessor