test - migrate to dapr clientconn

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

Author: mikeee

.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

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.

examples/workflow/main.go

@@ -46,8 +46,11 @@ func main() {
 		if err := wr.Start(); err != nil {
 			log.Fatal(err)
 		}
+		time.Sleep(time.Minute)
 	}()
 
+	time.Sleep(time.Second * 10)
+
 	daprClient, err := client.NewClient()
 	defer daprClient.Close()
 	if err != nil {
@@ -118,7 +121,7 @@ func main() {
 
 	fmt.Println("workflow resumed")
 
-	fmt.Printf("stage: %d", stage)
+	fmt.Printf("stage: %d\n", stage)
 
 	// Raise Event Test
 
@@ -226,8 +229,6 @@ func main() {
 	if err := wr.Shutdown(); err != nil {
 		log.Fatalf("failed to shutdown runtime: %v", err)
 	}
-
-	time.Sleep(time.Second * 5)
 }
 
 func TestWorkflow(ctx *workflow.Context) (any, error) {

workflow/runtime.go

@@ -2,32 +2,25 @@ package workflow
 
 import (
 	"context"
-	"crypto/tls"
 	"errors"
 	"fmt"
-	"github.com/dapr/go-sdk/workflow/internal"
-	"google.golang.org/grpc/credentials"
+	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)
@@ -39,28 +32,17 @@ func NewRuntime(address string) (*WorkflowRuntime, error) {
 		return &WorkflowRuntime{}, errors.New("no address provided")
 	}
 
-	ctx, cancel := context.WithTimeout(context.Background(), time.Second*5) // TODO: add timeout option
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
 
-	connectionOptions := []grpc.DialOption{
-		grpc.WithUserAgent(internal.UserAgent()),
-		grpc.WithBlock(),
-	}
+	daprClient, err := dapr.NewClientWithAddressContext(ctx, address)
 
-	parsedAddress, err := internal.ParseGRPCEndpoint(address)
-
-	if parsedAddress.TLS {
-		connectionOptions = append(connectionOptions, grpc.WithTransportCredentials(credentials.NewTLS(new(tls.Config))))
-	} else {
-		connectionOptions = append(connectionOptions, grpc.WithTransportCredentials(insecure.NewCredentials()))
+	if err != nil {
+		return &WorkflowRuntime{}, fmt.Errorf("failed to create runtime - dapr client failed: %v", err)
 	}
 
-	clientConn, err := grpc.DialContext(
-		ctx,
-		parsedAddress.Target,
-		connectionOptions...,
-	)
+	clientConn := daprClient.GrpcClientConn()
 
-	cancel()
 	if err != nil {
 		return &WorkflowRuntime{}, fmt.Errorf("failed to create runtime - grpc connection failed: %v", err)
 	}
@@ -69,7 +51,6 @@ func NewRuntime(address string) (*WorkflowRuntime, error) {
 		tasks:  task.NewTaskRegistry(),
 		client: client.NewTaskHubGrpcClient(clientConn, backend.DefaultLogger()),
 		quit:   make(chan bool),
-		cancel: cancel,
 	}, nil
 }
 
@@ -145,8 +126,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")

workflow/runtime_test.go

@@ -12,7 +12,7 @@ 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)
 	})
@@ -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