用 Go 提交任务

package main

import (
	"context"
	"flag"
	"fmt"
	"io"
	"log"
	"time"

	"github.com/google/uuid"
	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials/insecure"

	workerpb "github.com/Chaintable/blockx/api/grpc/worker/workerpb"
)

func main() {
	addr := flag.String("worker", "127.0.0.1:9221", "Worker address")
	flag.Parse()
	conn, err := grpc.NewClient("passthrough:///"+*addr,
		grpc.WithTransportCredentials(insecure.NewCredentials()))
	if err != nil {
		log.Fatal(err)
	}
	defer conn.Close()
	client := workerpb.NewWorkerServiceClient(conn)
	ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
	defer cancel()

	taskID := uuid.NewString()
	fmt.Println("task:", taskID)
	slot, err := client.RequestTaskSlot(ctx, &workerpb.RequestTaskSlotRequest{
		TaskId: taskID, TtlMs: 5000,
	})
	if err != nil {
		log.Fatal(err)
	}
	submit, err := client.SubmitTask(ctx, &workerpb.SubmitTaskRequest{
		SlotId: slot.GetSlotId(),
		Task: &workerpb.TaskInput{
			TaskId: taskID,
			FunctionCallConfig: &workerpb.FunctionCallConfig{
				Type:   "InputsCallConfig",
				Config: []byte(`{"function":"","code":{"sourceCode":"def _(name):\n    return {\"message\": f\"hello, {name}\"}\n"},"callList":[["ct2"],["blockx"]]}`),
			},
			ResultHandler: &workerpb.ResultHandler{Type: "ReturnValueHandler", Config: []byte(`{}`)},
			TaskTimeoutMs: 30000,
		},
	})
	if err != nil {
		log.Fatal(err)
	}
	fmt.Println("submitted:", submit.GetState())

	stream, err := client.WatchTasks(ctx, &workerpb.WatchTasksRequest{TaskIds: []string{taskID}})
	if err != nil {
		log.Fatal(err)
	}
	for {
		update, err := stream.Recv()
		if err == io.EOF {
			log.Fatal("watch ended before a terminal result; inspect task ", taskID)
		}
		if err != nil {
			log.Fatal(err)
		}
		if update.GetError() != "" {
			log.Fatal(update.GetError())
		}
		if !update.GetIsTerminal() {
			continue
		}
		result := update.GetResult()
		if result == nil {
			log.Fatal("terminal update has no result")
		}
		fmt.Println("state:", update.GetState())
		if !result.GetSuccess() {
			log.Fatalf("task failed: %s: %s", result.GetExecuteResult().GetFailureCode(),
				result.GetExecuteResult().GetFailureMessage())
		}
		for _, p := range result.GetExecuteResult().GetPluginResults() {
			fmt.Printf("%s success=%v result=%s\n", p.GetPluginName(), p.GetSuccess(), p.GetResult())
		}
		return
	}
}

go run /tmp/blockx-grpc-example.go -worker 127.0.0.1:9221