deliveryhero / pipeline Goto Github PK

func RunPipeline() (txinput.TxInput, error) {
         // create stakeDepositsNum of stakeDepositMessagesRequests.
	stakeDepositsNum := dp.Amount / uint64(s.maxStakedBalance)
	stakeDepositMessageRequest := s.createStakeDepositMessageRequest(authInfo, stake.ID, dp)
	stakeDepositMessagesRequests := s.createStakeDepositMessageRequests(stakeDepositMessageRequest, stakeDepositsNum)

	// 1. Create a pipeline that yields req.StakeDepositMessage stakeDepositsNum times.
	// 2. Process req.StakeDepositMessages and turn them into DepositMessageResult.
	// 3. Process DepositMessageResult and turn them into DepositTxInputResult.
	p := pipeline.Emit(stakeDepositMessagesRequests...)
	p = pipeline.ProcessConcurrently(ctx, s.numCPU, s.depositMessageProcessor, p)
	p = pipeline.ProcessConcurrently(ctx, s.numCPU, s.depositTxInputProcessor, p)

	// allocate results
	txInputs := make([]txinput.TxInput, stakeDepositsNum)
	
	// if range over p at this time I have no guarantee, that the pipeline goroutines have done their jobs.
        // there should be a way to wait for the pipeline to finish.
       for v := range p {
          // collect results  
       } 

       return txInputs, nil
}

func RunPipeline() (txinput.TxInput, error) {
	// allocate results
	txInputs := make([]txinput.TxInput, stakeDepositsNum)

	// init channels and an error group to sync goroutines
	depositMessageResultCh := make(chan processor.DepositMessageResult)
	depositTxInputResultCh := make(chan processor.DepositTxInputResult)

	g, gCtx := errgroup.WithContext(ctx)

	// 1. Create a pipeline that yields req.StakeDepositMessage stakeDepositsNum times.
	pipe := s.repeatFnTimes(gCtx, stakeDepositsNum, s.createStakeDepositMessageRequestFn(authInfo, stake.ID, dp))

	// 2. Process req.StakeDepositMessages and turn them into DepositMessageResult.
	g.Go(s.processDepositMessageRequests(gCtx, pipe, depositMessageResultCh))

	// 3. Process DepositMessageResult and turn them into DepositTxInputResult.
	g.Go(s.processDepositMessageResult(gCtx, depositMessageResultCh, depositTxInputResultCh))

	// 4. Collect TxInputs.
	g.Go(s.collect(gCtx, depositTxInputResultCh, txInputs))

	// wait for goroutines to finish.
	if err := g.Wait(); err != nil {
		return nil, merry.Wrap(ErrCannotRunStakingPipeline, merry.WithCause(err))
	}

        return txInputs, nil
}

type Result struct {
    Err error
    Value interface{}
}

func (p *SomeProcessor) Process(ctx context.Context, ins interface{}) (interface{}, error) {
	insSlice, err := toSliceOfInterfaces(ins)
	if err != nil {
		return nil, merry.Wrap(ErrCannotProcessDepositTxInput, merry.WithCause(err))
	}

	result := make([]interface{}, 0, len(insSlice))

	for _, in := range insSlice {
		res, err := ToResult(in)
		if err != nil {
			result = append(result, Result{
				Err:   errors.Wrap(ErrCannotConvertInterfaceToResult),
				Value: nil,
			})

			continue
		}

		foo, err := foo(res.Value)
		if err != nil {
			result = append(result, Result{
				Err:   errors.Wrap(ErrCannotRunFoo),
				Value: nil,
			})

			continue
		}

		bar, err := bar(ctx, foo)
		if err != nil {
			result = append(result, Result{
				Err:   errors.Wrap(ErrCannotRunBar),
				Value: nil,
			})

			continue
		}

		result = append(result, Result{Err: nil, Value: bar})
	}

	return result, nil
}