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
}
}