code: func (a *application) processMessage(c context.Context, coll *collector.Collector, msg *pubsub.Message) bool {
  	log.Fields{
  		"size": len(msg.Data),
  	}.Infof(c, "Received Pub/Sub Message.")
  
  	startTime := clock.Now(c)
  	err := coll.Process(c, msg.Data)
  	duration := clock.Now(c).Sub(startTime)
  
  	// We track processing time in milliseconds.
  	tsTaskProcessingTime.Add(c, duration.Seconds()*1000)
  
  	switch {
  	case transient.Tag.In(err):
  		// Do not consume
  		log.Fields{
  			log.ErrorKey: err,
  			"duration":   duration,
  		}.Warningf(c, "TRANSIENT error ingesting Pub/Sub message.")
  		tsPubsubCount.Add(c, 1, "transient_failure")
  		return false
  
  	case err == nil:
  		log.Fields{
  			"size":     len(msg.Data),
  			"duration": duration,
  		}.Infof(c, "Message successfully processed; ACKing.")
  		tsPubsubCount.Add(c, 1, "success")
  		return true
  
  	default:
  		log.Fields{
  			log.ErrorKey: err,
  			"size":       len(msg.Data),
  			"duration":   duration,
  		}.Errorf(c, "Non-transient error ingesting Pub/Sub message; ACKing.")
  		tsPubsubCount.Add(c, 1, "failure")
  		return true
  	}
  }
docs: // processMessage returns true if the message should be ACK'd (deleted from
  // Pub/Sub) or false if the message should not be ACK'd.