// Jobs creates a new consumer on the queue, and returns three channels. The
// first channel gets sent every BuildJob that we receive from AMQP. The
// stopChan is a channel that can be closed in order to stop the consumer.
Jobs creates a new consumer on the queue, and returns three channels. The
first channel gets sent every BuildJob that we receive from AMQP. The
stopChan is a channel that can be closed in order to stop the consumer.
// Jobs creates a new consumer on the queue and returns three channels . The // first channel gets sent every BuildJob that we receive from AMQP . The // stopChan is a channel that can be closed in order to stop the consumer .
func (q *AMQPJobQueue) Jobs(ctx gocontext.Context) (outChan <-chan Job, err error) {
	jobsChannel, err := q.conn.Channel()
	if err != nil {
		return
	}
	err = jobsChannel.Qos(1, 0, false)
	if err != nil {
		return
	}
	deliveries, err := jobsChannel.Consume(
		q.queue,              
		"build-job-consumer", 
		false, 
		false, 
		false, 
		false, 
		amqp.Table{"x-priority": int64(q.priority)}) 
	if err != nil {
		return
	}
	logWriterChannel, err := q.conn.Channel()
	if err != nil {
		return
	}
	buildJobChan := make(chan Job)
	outChan = buildJobChan
	go func() {
		defer jobsChannel.Close()
		defer logWriterChannel.Close()
		defer close(buildJobChan)
		logger := context.LoggerFromContext(ctx).WithFields(logrus.Fields{
			"self": "amqp_job_queue",
			"inst": fmt.Sprintf("%p", q),
		})
		for {
			if ctx.Err() != nil {
				return
			}
			select {
			case <-ctx.Done():
				return
			case <-time.After(time.Second):
				continue
			case delivery, ok := <-deliveries:
				if !ok {
					logger.Info("job queue channel closed")
					return
				}
				buildJob := &amqpJob{
					payload:         &JobPayload{},
					startAttributes: &backend.StartAttributes{},
					stateUpdatePool: q.stateUpdatePool,
					withLogSharding: q.withLogSharding,
					logWriterChan:   logWriterChannel,
					delivery:        delivery,
				}
				startAttrs := &jobPayloadStartAttrs{Config: &backend.StartAttributes{}}
				err := json.Unmarshal(delivery.Body, buildJob.payload)
				if err != nil {
					logger.WithField("err", err).Error("payload JSON parse error, attempting to ack+drop delivery")
					err := delivery.Ack(false)
					if err != nil {
						logger.WithField("err", err).WithField("delivery", delivery).Error("couldn't ack+drop delivery")
					}
					continue
				}
				logger.WithField("job_id", buildJob.payload.Job.ID).Info("received amqp delivery")
				err = json.Unmarshal(delivery.Body, &startAttrs)
				if err != nil {
					logger.WithField("err", err).Error("start attributes JSON parse error, attempting to ack+drop delivery")
					err := delivery.Ack(false)
					if err != nil {
						logger.WithField("err", err).WithField("delivery", delivery).Error("couldn't ack+drop delivery")
					}
					err = buildJob.Error(ctx, "An error occured while parsing the job config. Please consider enabling the build config validation feature for the repository: https://docs.travis-ci.com/user/build-config-validation")
					if err != nil {
						logger.WithField("err", err).Error("couldn't error the job")
					}
					continue
				}
				buildJob.rawPayload, err = simplejson.NewJson(delivery.Body)
				if err != nil {
					logger.WithField("err", err).Error("raw payload JSON parse error, attempting to ack+drop delivery")
					err := delivery.Ack(false)
					if err != nil {
						logger.WithField("err", err).WithField("delivery", delivery).Error("couldn't ack+drop delivery")
					}
					continue
				}
				buildJob.startAttributes = startAttrs.Config
				buildJob.startAttributes.VMType = buildJob.payload.VMType
				buildJob.startAttributes.VMSize = buildJob.payload.VMSize
				buildJob.startAttributes.VMConfig = buildJob.payload.VMConfig
				buildJob.startAttributes.Warmer = buildJob.payload.Warmer
				buildJob.startAttributes.SetDefaults(q.DefaultLanguage, q.DefaultDist, q.DefaultArch, q.DefaultGroup, q.DefaultOS, VMTypeDefault, VMConfigDefault)
				buildJob.conn = q.conn
				buildJob.stateCount = buildJob.payload.Meta.StateUpdateCount
				jobSendBegin := time.Now()
				select {
				case buildJobChan <- buildJob:
					metrics.TimeSince("travis.worker.job_queue.amqp.blocking_time", jobSendBegin)
					logger.WithFields(logrus.Fields{
						"source":           "amqp",
						"send_duration_ms": time.Since(jobSendBegin).Seconds() * 1e3,
					}).Info("sent job to output channel")
				case <-ctx.Done():
					_ = delivery.Nack(false, true)
					return
				}
			}
		}
	}()
	return
}
func ( q * AMQPJobQueue ) Jobs ( ctx gocontext . Context ) ( outChan <- chan Job , err error ) { jobsChannel , err := q . conn . Channel ( ) if err != nil { return } err = jobsChannel . Qos ( 1 , 0 , false ) if err != nil { return } deliveries , err := jobsChannel . Consume ( q . queue , " " , false , false , false , false , amqp . Table { " " : int64 ( q . priority ) } ) if err != nil { return } logWriterChannel , err := q . conn . Channel ( ) if err != nil { return } buildJobChan := make ( chan Job ) outChan = buildJobChan go func ( ) { defer jobsChannel . Close ( ) defer logWriterChannel . Close ( ) defer close ( buildJobChan ) logger := context . LoggerFromContext ( ctx ) . WithFields ( logrus . Fields { " " : " " , " " : fmt . Sprintf ( " " , q ) , } ) for { if ctx . Err ( ) != nil { return } select { case <- ctx . Done ( ) : return case <- time . After ( time . Second ) : continue case delivery , ok := <- deliveries : if ! ok { logger . Info ( " " ) return } buildJob := & amqpJob { payload : & JobPayload { } , startAttributes : & backend . StartAttributes { } , stateUpdatePool : q . stateUpdatePool , withLogSharding : q . withLogSharding , logWriterChan : logWriterChannel , delivery : delivery , } startAttrs := & jobPayloadStartAttrs { Config : & backend . StartAttributes { } } err := json . Unmarshal ( delivery . Body , buildJob . payload ) if err != nil { logger . WithField ( " " , err ) . Error ( " " ) err := delivery . Ack ( false ) if err != nil { logger . WithField ( " " , err ) . WithField ( " " , delivery ) . Error ( " " ) } continue } logger . WithField ( " " , buildJob . payload . Job . ID ) . Info ( " " ) err = json . Unmarshal ( delivery . Body , & startAttrs ) if err != nil { logger . WithField ( " " , err ) . Error ( " " ) err := delivery . Ack ( false ) if err != nil { logger . WithField ( " " , err ) . WithField ( " " , delivery ) . Error ( " " ) } err = buildJob . Error ( ctx , " " ) if err != nil { logger . WithField ( " " , err ) . Error ( " " ) } continue } buildJob . rawPayload , err = simplejson . NewJson ( delivery . Body ) if err != nil { logger . WithField ( " " , err ) . Error ( " " ) err := delivery . Ack ( false ) if err != nil { logger . WithField ( " " , err ) . WithField ( " " , delivery ) . Error ( " " ) } continue } buildJob . startAttributes = startAttrs . Config buildJob . startAttributes . VMType = buildJob . payload . VMType buildJob . startAttributes . VMSize = buildJob . payload . VMSize buildJob . startAttributes . VMConfig = buildJob . payload . VMConfig buildJob . startAttributes . Warmer = buildJob . payload . Warmer buildJob . startAttributes . SetDefaults ( q . DefaultLanguage , q . DefaultDist , q . DefaultArch , q . DefaultGroup , q . DefaultOS , VMTypeDefault , VMConfigDefault ) buildJob . conn = q . conn buildJob . stateCount = buildJob . payload . Meta . StateUpdateCount jobSendBegin := time . Now ( ) select { case buildJobChan <- buildJob : metrics . TimeSince ( " " , jobSendBegin ) logger . WithFields ( logrus . Fields { " " : " " , " " : time . Since ( jobSendBegin ) . Seconds ( ) * 1e3 , } ) . Info ( " " ) case <- ctx . Done ( ) : _ = delivery . Nack ( false , true ) return } } } } ( ) return }
Jobs creates a new consumer on the queue, and returns three channels.
Jobs creates a new consumer on the queue and returns three channels .
// queue // consumer // autoAck // exclusive // noLocal // noWait // args