fix: migrate to dapr builtin client

Signed-off-by: mikeee <[email protected]>

Author: mikeee

Diff for: .github/workflows/validate_examples.yaml

@@ -92,11 +92,4 @@ jobs:
       - name: Check Examples
         run: |
           cd examples
-          ./validate.sh actor
-          ./validate.sh configuration
-          ./validate.sh grpc-service
-          ./validate.sh hello-world
-          ./validate.sh pubsub
-          ./validate.sh service
-          ./validate.sh socket
           ./validate.sh workflow

Diff for: client/client.go

@@ -253,6 +253,8 @@ type Client interface {
 
 	// GrpcClient returns the base grpc client if grpc is used and nil otherwise
 	GrpcClient() pb.DaprClient
+
+	GrpcClientConn() *grpc.ClientConn
 }
 
 // NewClient instantiates Dapr client using DAPR_GRPC_PORT environment variable as port.

Diff for: examples/workflow/README.md

@@ -19,6 +19,7 @@ expected_stdout_lines:
   - '== APP == workflow started with id: a7a4168d-3a1c-41da-8a4f-e7f6d9c718d9'
   - '== APP == workflow paused'
   - '== APP == workflow resumed'
+  - '== APP == stage: 1'
   - '== APP == workflow event raised'
   - '== APP == stage: 2'
   - '== APP == workflow status: COMPLETED'
@@ -33,7 +34,6 @@ sleep: 30
 
 ```bash
 dapr run --app-id workflow-sequential \
-         --app-protocol grpc \
          --dapr-grpc-port 50001 \
          --placement-host-address localhost:50005 \
          --log-level debug \
@@ -55,6 +55,7 @@ dapr run --app-id workflow-sequential \
   - '== APP == workflow started with id: a7a4168d-3a1c-41da-8a4f-e7f6d9c718d9'
   - '== APP == workflow paused'
   - '== APP == workflow resumed'
+  - '== APP == stage: 1'
   - '== APP == workflow event raised'
   - '== APP == stage: 2'
   - '== APP == workflow status: COMPLETED'

Diff for: examples/workflow/main.go

@@ -19,7 +19,7 @@ const (
 )
 
 func main() {
-	wr, err := workflow.NewRuntime("localhost", "50001")
+	wr, err := workflow.NewRuntime("dns:127.0.0.1:50001")
 	if err != nil {
 		log.Fatal(err)
 	}
@@ -48,6 +48,8 @@ func main() {
 		}
 	}()
 
+	time.Sleep(time.Second * 5)
+
 	daprClient, err := client.NewClient()
 	defer daprClient.Close()
 	if err != nil {
@@ -116,7 +118,9 @@ func main() {
 		log.Fatalf("workflow not running")
 	}
 
-	fmt.Printf("workflow resumed\n")
+	fmt.Println("workflow resumed")
+
+	fmt.Printf("stage: %d", stage)
 
 	// Raise Event Test
 

Diff for: workflow/runtime.go

@@ -4,45 +4,42 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	dapr "github.com/dapr/go-sdk/client"
+	"github.com/microsoft/durabletask-go/backend"
+	"github.com/microsoft/durabletask-go/client"
+	"github.com/microsoft/durabletask-go/task"
 	"log"
 	"reflect"
 	"runtime"
 	"strings"
 	"sync"
-	"time"
-
-	"github.com/microsoft/durabletask-go/backend"
-	"github.com/microsoft/durabletask-go/client"
-	"github.com/microsoft/durabletask-go/task"
-	"google.golang.org/grpc"
-	"google.golang.org/grpc/credentials/insecure"
 )
 
 type WorkflowRuntime struct {
 	tasks  *task.TaskRegistry
 	client *client.TaskHubGrpcClient
 
-	mutex  sync.Mutex // TODO: implement
-	quit   chan bool
-	cancel context.CancelFunc
+	mutex sync.Mutex // TODO: implement
+	quit  chan bool
 }
 
 type Workflow func(ctx *Context) (any, error)
 
 type Activity func(ctx ActivityContext) (any, error)
 
-func NewRuntime(host string, port string) (*WorkflowRuntime, error) {
-	ctx, cancel := context.WithTimeout(context.Background(), time.Second*10) // TODO: add timeout option
-	defer cancel()
+func NewRuntime(address string) (*WorkflowRuntime, error) {
+	if address == "" {
+		return &WorkflowRuntime{}, errors.New("no address provided")
+	}
 
-	address := fmt.Sprintf("%s:%s", host, port)
+	ctx, cancel := context.WithCancel(context.Background()) // TODO: add timeout option
+	defer cancel()
+	daprClient, err := dapr.NewClientWithAddressContext(ctx, address)
+	if err != nil {
+		return nil, err
+	}
 
-	clientConn, err := grpc.DialContext(
-		ctx,
-		address,
-		grpc.WithTransportCredentials(insecure.NewCredentials()),
-		grpc.WithBlock(), // TODO: config
-	)
+	clientConn := daprClient.GrpcClientConn()
 	if err != nil {
 		return &WorkflowRuntime{}, fmt.Errorf("failed to create runtime - grpc connection failed: %v", err)
 	}
@@ -51,7 +48,6 @@ func NewRuntime(host string, port string) (*WorkflowRuntime, error) {
 		tasks:  task.NewTaskRegistry(),
 		client: client.NewTaskHubGrpcClient(clientConn, backend.DefaultLogger()),
 		quit:   make(chan bool),
-		cancel: cancel,
 	}, nil
 }
 
@@ -127,8 +123,6 @@ func (wr *WorkflowRuntime) Start() error {
 }
 
 func (wr *WorkflowRuntime) Shutdown() error {
-	// cancel grpc context
-	wr.cancel()
 	// send close signal
 	wr.quit <- true
 	log.Println("work item listener shutdown signal sent")

Diff for: workflow/runtime_test.go

@@ -12,9 +12,9 @@ import (
 
 func TestNewRuntime(t *testing.T) {
 	t.Run("failure to create newruntime without dapr", func(t *testing.T) {
-		wr, err := NewRuntime("localhost", "50001")
+		wr, err := NewRuntime(":50001")
 		require.Error(t, err)
-		assert.Equal(t, &WorkflowRuntime{}, wr)
+		assert.Empty(t, wr)
 	})
 }
 
@@ -24,7 +24,6 @@ func TestWorkflowRuntime(t *testing.T) {
 		client: nil,
 		mutex:  sync.Mutex{},
 		quit:   nil,
-		cancel: nil,
 	}
 
 	// TODO: Mock grpc conn - currently requires dapr to be available