Function runOutputLoop

pkg/jobcontroller/jobcontroller.go:814–862

(ctx context.Context, jobId string, streamId string, reader *streamclient.Reader)

}

func runOutputLoop(ctx context.Context, jobId string, streamId string, reader *streamclient.Reader) {
	defer reader.Close()
	defer func() {
		log.Printf("[job:%s] [stream:%s] output loop finished", jobId, streamId)
	}()

	log.Printf("[job:%s] [stream:%s] output loop started", jobId, streamId)
	buf := make([]byte, 4096)
	for {
		n, err := reader.Read(buf)
		currentStreamId, _ := jobStreamIds.GetEx(jobId)
		if currentStreamId != streamId {
			log.Printf("[job:%s] [stream:%s] stream superseded by [stream:%s], exiting output loop", jobId, streamId, currentStreamId)
			break
		}
		if n > 0 {
			appendErr := handleAppendJobFile(ctx, jobId, JobOutputFileName, buf[:n])
			if appendErr != nil {
				log.Printf("[job:%s] error appending data to WaveFS: %v", jobId, appendErr)
			}
		}

		if err == io.EOF {
			log.Printf("[job:%s] stream ended (EOF)", jobId)
			updateErr := wstore.DBUpdateFn(ctx, jobId, func(job *waveobj.Job) {
				job.StreamDone = true
			})
			if updateErr != nil {
				log.Printf("[job:%s] error updating job stream status: %v", jobId, updateErr)
			}
			tryTerminateJobManager(ctx, jobId)
			break
		}

		if err != nil {
			log.Printf("[job:%s] stream error: %v", jobId, err)
			streamErr := err.Error()
			updateErr := wstore.DBUpdateFn(ctx, jobId, func(job *waveobj.Job) {
				job.StreamDone = true
				job.StreamError = streamErr
			})
			if updateErr != nil {
				log.Printf("[job:%s] error updating job stream error: %v", jobId, updateErr)
			}
			tryTerminateJobManager(ctx, jobId)
			break
		}
	}
}

func HandleCmdJobExited(ctx context.Context, jobId string, data wshrpc.CommandJobCmdExitedData) error {
	var updatedJob *waveobj.Job