Merge branch 'master' of https://gitlink.org.cn/cloudream/common

pkgs/future/future.go

@@ -5,7 +5,11 @@ import (
 	"fmt"
 )
 
-var ErrCompleted = fmt.Errorf("context canceled")
+var ErrConsumed = fmt.Errorf("future already consumed")
+
+var ErrNotComplete = fmt.Errorf("future not complete")
+
+var ErrCanceled = context.Canceled
 
 type Future interface {
 	IsComplete() bool

pkgs/future/ready.go

@@ -24,7 +24,7 @@ func (f *Ready) Wait(ctx context.Context) error {
 	select {
 	case v, ok := <-f.ch:
 		if !ok {
-			return ErrCompleted
+			return ErrConsumed
 		}
 		return v
 

pkgs/future/set_value_future.go

@@ -63,7 +63,7 @@ func (f *SetValueFuture[T]) Wait(ctx context.Context) (T, error) {
 	case cv, ok := <-f.ch:
 		if !ok {
 			var ret T
-			return ret, cv.Err
+			return ret, ErrConsumed
 		}
 		return cv.Value, cv.Err
 
@@ -73,6 +73,20 @@ func (f *SetValueFuture[T]) Wait(ctx context.Context) (T, error) {
 	}
 }
 
+func (f *SetValueFuture[T]) TryGetValue() (T, error) {
+	select {
+	case cv, ok := <-f.ch:
+		if !ok {
+			var ret T
+			return ret, ErrConsumed
+		}
+		return cv.Value, cv.Err
+	default:
+		var ret T
+		return ret, ErrNotComplete
+	}
+}
+
 type SetValueFuture2[T1 any, T2 any] struct {
 	isCompleted  bool
 	ch           chan ChanValue2[T1, T2]
@@ -126,7 +140,7 @@ func (f *SetValueFuture2[T1, T2]) Wait(ctx context.Context) (T1, T2, error) {
 	select {
 	case cv, ok := <-f.ch:
 		if !ok {
-			return cv.Value1, cv.Value2, cv.Err
+			return cv.Value1, cv.Value2, ErrConsumed
 		}
 		return cv.Value1, cv.Value2, cv.Err
 

pkgs/future/set_void_future.go

@@ -41,7 +41,7 @@ func (f *SetVoidFuture) Wait(ctx context.Context) error {
 	select {
 	case v, ok := <-f.ch:
 		if !ok {
-			return ErrCompleted
+			return ErrConsumed
 		}
 		return v
 

pkgs/ioswitch/plan/ops/var.go

@@ -1,6 +1,7 @@
 package ops
 
 import (
+	"gitlink.org.cn/cloudream/common/pkgs/ioswitch/dag"
 	"gitlink.org.cn/cloudream/common/pkgs/ioswitch/exec"
 )
 
@@ -21,3 +22,31 @@ func (o *ConstVar) Execute(ctx *exec.ExecContext, e *exec.Executor) error {
 func (o *ConstVar) String() string {
 	return "ConstVar"
 }
+
+type ConstNode struct {
+	dag.NodeBase
+	Value exec.VarValue
+}
+
+func (b *GraphNodeBuilder) NewConst(val exec.VarValue) *ConstNode {
+	node := &ConstNode{
+		Value: val,
+	}
+	b.AddNode(node)
+
+	node.OutputValues().Init(node, 1)
+	return node
+}
+
+func (t *ConstNode) Output() dag.ValueOutputSlot {
+	return dag.ValueOutputSlot{
+		Node:  t,
+		Index: 0,
+	}
+}
+
+func (t *ConstNode) GenerateOp() (exec.Op, error) {
+	return &DropStream{
+		Input: t.Output().Var().VarID,
+	}, nil
+}