<commit_before>package calcium

import (
	"bufio"
	"context"
	"io"
	"strconv"
	"sync"

	"github.com/google/uuid"
	"github.com/pkg/errors"

	enginetypes "github.com/projecteru2/core/engine/types"
	"github.com/projecteru2/core/log"
	"github.com/projecteru2/core/strategy"
	"github.com/projecteru2/core/types"
	"github.com/projecteru2/core/utils"
	"github.com/projecteru2/core/wal"
)

const (
	exitDataPrefix = "[exitcode] "
	labelLambdaID  = "LambdaID"
)

// RunAndWait implement lambda
func (c *Calcium) RunAndWait(ctx context.Context, opts *types.DeployOptions, inCh <-chan []byte) (<-chan *types.AttachWorkloadMessage, error) {
	logger := log.WithField("Calcium", "RunAndWait").WithField("opts", opts)
	if err := opts.Validate(); err != nil {
		return nil, logger.Err(err)
	}
	opts.Lambda = true
	// count = 1 && OpenStdin
	if opts.OpenStdin && (opts.Count != 1 || opts.DeployStrategy != strategy.Auto) {
		logger.Errorf("Count %d method %s", opts.Count, opts.DeployStrategy)
		return nil, errors.WithStack(types.ErrRunAndWaitCountOneWithStdin)
	}

	commit, err := c.walCreateLambda(ctx, opts)
	if err != nil {
		return nil, logger.Err(err)
	}
	createChan, err := c.CreateWorkload(ctx, opts)
	if err != nil {
		logger.Errorf("[RunAndWait] Create workload error %+v", err)
		return nil, err
	}

	runMsgCh := make(chan *types.AttachWorkloadMessage)
	wg := &sync.WaitGroup{}
	for message := range createChan {
		if message.Error != nil || message.WorkloadID == "" {
			logger.Errorf("[RunAndWait] Create workload failed %+v", message.Error)
			continue
		}<commit_msg>workload id will be returned as the first message (#364)<commit_after>package calcium

import (
	"bufio"
	"context"
	"io"
	"strconv"
	"sync"

	"github.com/google/uuid"
	"github.com/pkg/errors"

	enginetypes "github.com/projecteru2/core/engine/types"
	"github.com/projecteru2/core/log"
	"github.com/projecteru2/core/strategy"
	"github.com/projecteru2/core/types"
	"github.com/projecteru2/core/utils"
	"github.com/projecteru2/core/wal"
)

const (
	exitDataPrefix = "[exitcode] "
	labelLambdaID  = "LambdaID"
)

// RunAndWait implement lambda
func (c *Calcium) RunAndWait(ctx context.Context, opts *types.DeployOptions, inCh <-chan []byte) (<-chan *types.AttachWorkloadMessage, error) {
	logger := log.WithField("Calcium", "RunAndWait").WithField("opts", opts)
	if err := opts.Validate(); err != nil {
		return nil, logger.Err(err)
	}
	opts.Lambda = true
	// count = 1 && OpenStdin
	if opts.OpenStdin && (opts.Count != 1 || opts.DeployStrategy != strategy.Auto) {
		logger.Errorf("Count %d method %s", opts.Count, opts.DeployStrategy)
		return nil, errors.WithStack(types.ErrRunAndWaitCountOneWithStdin)
	}

	commit, err := c.walCreateLambda(ctx, opts)
	if err != nil {
		return nil, logger.Err(err)
	}
	createChan, err := c.CreateWorkload(ctx, opts)
	if err != nil {
		logger.Errorf("[RunAndWait] Create workload error %+v", err)
		return nil, err
	}

	runMsgCh := make(chan *types.AttachWorkloadMessage)
	wg := &sync.WaitGroup{}
	for message := range createChan {
		if message.Error != nil || message.WorkloadID == "" {
			logger.Errorf("[RunAndWait] Create workload failed %+v", message.Error)
			continue
		}

		// first message to indicate the workload id
		runMsgCh <- &types.AttachWorkloadMessage{
			WorkloadID:    message.WorkloadID,
			Data:          []byte(""),
			StdStreamType: types.Stdout,
		}
{'fraction_non_alphanumeric': 0.149807178878671, 'fraction_numerical': 0.005636309700385643, 'mean_word_length': 4.386581469648562, 'pattern_counts': {'":': 0, '<': 8, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2851126', 'n_tokens_mistral': 1271, 'n_tokens_neox': 1203, 'n_words': 256}