Function restartStreaming

pkg/jobcontroller/jobcontroller.go:1204–1345

(ctx context.Context, jobId string, knownConnected bool, rtOpts *waveobj.RuntimeOpts)

}

func restartStreaming(ctx context.Context, jobId string, knownConnected bool, rtOpts *waveobj.RuntimeOpts) error {
	job, err := wstore.DBMustGet[*waveobj.Job](ctx, jobId)
	if err != nil {
		return fmt.Errorf("failed to get job: %w", err)
	}

	termSize := job.CmdTermSize
	if rtOpts != nil && rtOpts.TermSize.Rows > 0 && rtOpts.TermSize.Cols > 0 {
		termSize = rtOpts.TermSize
		err = wstore.DBUpdateFn(ctx, jobId, func(job *waveobj.Job) {
			job.CmdTermSize = termSize
		})
		if err != nil {
			log.Printf("[job:%s] warning: failed to update termsize in DB: %v", jobId, err)
		}
	}

	if !knownConnected {
		isConnected, err := conncontroller.IsConnected(job.Connection)
		if err != nil {
			return fmt.Errorf("error checking connection status: %w", err)
		}
		if !isConnected {
			return fmt.Errorf("connection %q is not connected", job.Connection)
		}

		jobConnStatus := GetJobConnStatus(jobId)
		if jobConnStatus != JobConnStatus_Connected {
			return fmt.Errorf("job manager is not connected (status: %s)", jobConnStatus)
		}
	}

	var currentSeq int64 = 0
	var totalGap int64 = 0
	waveFile, err := filestore.WFS.Stat(ctx, jobId, JobOutputFileName)
	if err == nil {
		currentSeq = waveFile.Size
		totalGap = getMetaInt64(waveFile.Meta, MetaKey_TotalGap)
		currentSeq += totalGap
	}

	bareRpc := wshclient.GetBareRpcClient()
	broker := bareRpc.StreamBroker
	readerRouteId := wshclient.GetBareRpcClientRouteId()
	writerRouteId := wshutil.MakeJobRouteId(jobId)
	reader, streamMeta := broker.CreateStreamReaderWithSeq(readerRouteId, writerRouteId, DefaultStreamRwnd, currentSeq)
	jobStreamIds.Set(jobId, streamMeta.Id)

	prepareData := wshrpc.CommandJobPrepareConnectData{
		StreamMeta: *streamMeta,
		Seq:        currentSeq,
		TermSize:   termSize,
	}

	rpcOpts := &wshrpc.RpcOpts{
		Route:   wshutil.MakeJobRouteId(jobId),
		Timeout: 5000,
	}