内置包装器

  	stopCh := make(chan struct{})

	p := ograph.NewPipeline()

	e := ograph.NewElement("node_to_async_run").
		UseFn(func() error {
			time.Sleep(time.Second)
			fmt.Println("node end")
			stopCh <- struct{}{}
			return nil
		}).
		Wrap(ogimpl.Async)

	// async node wont't block pipeline
	p.Register(e).Run(context.TODO(), nil)

	// The pipeline will end immediately after completion,
	// while the node will end one second after startup.
	fmt.Println("pipeline end")
	<-stopCh
  

  type ReadWriteStateNode struct {
	ograph.BaseNode
}

func (node *ReadWriteStateNode) Run(ctx context.Context, state ogcore.State) error {
	state.Set("i", 1)
	state.Update("i", func(val any) any {
		return "1"
	})
	state.Get("i")
	return nil
}
  

  	p := ograph.NewPipeline()

	e := ograph.NewElement("node_to_debug").
		UseNode(&ReadWriteStateNode{}).
		Wrap(ogimpl.Debug)

	// The state of a node before and after reading and writing will be recorded in the log.
	p.Register(e).Run(context.TODO(), nil)
  

  	p := ograph.NewPipeline()

	e := ograph.NewElement("node_to_delay").
		UseFn(func() error {
			fmt.Println("node start running")
			return nil
		}).
		Wrap(ogimpl.Delay).
		Params("Wait", "3s")

	// node_to_delay will run after 3 seconds
	p.Register(e).Run(context.TODO(), nil)
  

  	p := ograph.NewPipeline()

	e := ograph.NewElement("node_to_loop").
		UseFn(func() error {
			fmt.Println("node start running")
			return nil
		}).
		Wrap(ogimpl.Loop).
		Params("LoopTimes", 3)

	// node_to_loop will loop run 3 times
	p.Register(e).Run(context.TODO(), nil)
  

  	var once sync.Once

	p := ograph.NewPipeline()

	e := ograph.NewElement("problem_node").
		UseFn(func() error {
			var err error

			fmt.Println("node start running")

			once.Do(func() { // Returns error only once.
				err = errors.New("something going wrong")
			})

			return err
		}).
		Wrap(ogimpl.Retry)

	// The pipeline completes normally after the retry fails.
	err := p.Register(e).Run(context.TODO(), nil)
	fmt.Println(err == nil)
  

  	p := ograph.NewPipeline()

	e := ograph.NewElement("problem_node").
		UseFn(func() error {
			return errors.New("something going wrong")
		}).
		Wrap(ogimpl.Silent)

	// Pipeline will not fail due to failed nodes in the problem.
	err := p.Register(e).Run(context.TODO(), nil)
	fmt.Println(err == nil)
  

  	p := ograph.NewPipeline()

	e := ograph.NewElement("slow_node").
		UseFn(func() error {
			time.Sleep(time.Minute)
			return nil
		}).
		Wrap(ogimpl.Timeout).
		Params("Timeout", "10ms")

	// After a node times out, the pipeline stops running and returns a timeout error.
	err := p.Register(e).Run(context.TODO(), nil)
	fmt.Println(errors.Is(err, ogimpl.ErrTimeout))
  

  	p := ograph.NewPipeline()

	e := ograph.NewElement("node_to_be_trace").
		UseNode(&ograph.BaseNode{}).
		Wrap(ogimpl.Trace)

	// The execution process of the node will be recorded in the log.
	p.Register(e).Run(context.TODO(), nil)