Function SetupJobManager

pkg/jobmanager/jobmanager.go:48–94

(clientId string, jobId string, publicKeyBytes []byte, jobAuthToken string, readyFile *os.File)

}

func SetupJobManager(clientId string, jobId string, publicKeyBytes []byte, jobAuthToken string, readyFile *os.File) error {
	if runtime.GOOS != "linux" && runtime.GOOS != "darwin" {
		return fmt.Errorf("job manager only supported on unix systems, not %s", runtime.GOOS)
	}
	WshCmdJobManager.ClientId = clientId
	WshCmdJobManager.JobId = jobId
	WshCmdJobManager.JwtPublicKey = publicKeyBytes
	WshCmdJobManager.JobAuthToken = jobAuthToken
	WshCmdJobManager.StreamManager = MakeStreamManager()
	WshCmdJobManager.InputQueue = utilds.MakeQuickReorderQueue[wshrpc.CommandJobInputData](JobInputQueueSize, JobInputQueueTimeout)
	err := wavejwt.SetPublicKey(publicKeyBytes)
	if err != nil {
		return fmt.Errorf("failed to set public key: %w", err)
	}
	err = MakeJobDomainSocket(clientId, jobId)
	if err != nil {
		return err
	}

	go func() {
		defer func() {
			panichandler.PanicHandler("JobManager:processInputQueue", recover())
		}()
		WshCmdJobManager.processInputQueue()
	}()

	fmt.Fprintf(readyFile, JobManagerStartLabel+"\n")
	readyFile.Close()

	err = daemonize(clientId, jobId)
	if err != nil {
		return fmt.Errorf("failed to daemonize: %w", err)
	}

	go func() {
		defer func() {
			panichandler.PanicHandler("JobManager:keepalive", recover())
		}()
		ticker := time.NewTicker(1 * time.Hour)
		defer ticker.Stop()
		for range ticker.C {
			log.Printf("keepalive: job manager active\n")
		}
	}()

	return nil
}

func (jm *JobManager) processInputQueue() {
	for data := range jm.InputQueue.C() {