Expose workflow client from Activity

activity/activity.go

@@ -26,7 +26,7 @@ package activity
 
 import (
 	"context"
-
+	"go.temporal.io/sdk/client"
 	"go.temporal.io/sdk/internal"
 	"go.temporal.io/sdk/internal/common/metrics"
 	"go.temporal.io/sdk/log"
@@ -108,3 +108,9 @@ func GetWorkerStopChannel(ctx context.Context) <-chan struct{} {
 func IsActivity(ctx context.Context) bool {
 	return internal.IsActivity(ctx)
 }
+
+// GetClient returns a client that can be used to interact with the Temporal
+// service from an activity.
+func GetClient(ctx context.Context) client.Client {
+	return internal.GetClient(ctx)
+}

internal/activity.go

@@ -264,6 +264,13 @@ func RecordActivityHeartbeat(ctx context.Context, details ...interface{}) {
 	getActivityOutboundInterceptor(ctx).RecordHeartbeat(ctx, details...)
 }
 
+// GetClient returns a client that can be used to interact with the Temporal
+// service from an activity.
+func GetClient(ctx context.Context) Client {
+	activityEnv := getActivityEnv(ctx)
+	return activityEnv.serviceInvoker.GetClient(ClientOptions{Namespace: activityEnv.workflowNamespace})
+}
+
 // ServiceInvoker abstracts calls to the Temporal service from an activity implementation.
 // Implement to unit test activities.
 type ServiceInvoker interface {

internal/activity_test.go

@@ -249,3 +249,12 @@ func (s *activityTestSuite) TestIsActivity() {
 	ctx, _ = newActivityContext(context.Background(), nil, &activityEnvironment{workerStopChannel: ch})
 	s.True(IsActivity(ctx))
 }
+
+func (s *activityTestSuite) TestGetClient() {
+	ctx, cancel := context.WithCancel(context.Background())
+	invoker := newServiceInvoker([]byte("task-token"), "identity", s.service, metrics.NopHandler, cancel,
+		1*time.Second, make(chan struct{}), s.namespace)
+	ctx, _ = newActivityContext(ctx, nil, &activityEnvironment{serviceInvoker: invoker})
+	client := GetClient(ctx)
+	s.NotNil(client)
+}

test/activity_test.go

@@ -28,6 +28,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"go.temporal.io/api/workflowservice/v1"
 	"strconv"
 	"strings"
 	"sync"
@@ -435,3 +436,18 @@ func (a *Activities) register(worker worker.Worker) {
 	worker.RegisterActivityWithOptions(a.activities2, activity.RegisterOptions{Name: "Prefix_", DisableAlreadyRegisteredCheck: true})
 	worker.RegisterActivityWithOptions(a.InspectActivityInfo, activity.RegisterOptions{Name: "inspectActivityInfo"})
 }
+
+func (a *Activities) ClientFromActivity(ctx context.Context) error {
+	activityClient := activity.GetClient(ctx)
+	info := activity.GetInfo(ctx)
+	request := workflowservice.ListWorkflowExecutionsRequest{Namespace: info.WorkflowNamespace}
+	resp, err := activityClient.ListWorkflow(ctx, &request)
+	if err != nil {
+		return err
+	}
+
+	if len(resp.Executions) == 0 {
+		return fmt.Errorf("expected non-empty list of executions")
+	}
+	return nil
+}

test/integration_test.go

@@ -6633,6 +6633,11 @@ func (ts *IntegrationTestSuite) TestAwaitWithOptionsTimeout() {
 	ts.Equal("await-timer", str)
 }
 
+func (ts *IntegrationTestSuite) TestClientFromActivity() {
+	err := ts.executeWorkflow("client-from-activity", ts.workflows.WorkflowClientFromActivity, nil)
+	ts.NoError(err)
+}
+
 // executeWorkflow executes a given workflow and waits for the result
 func (ts *IntegrationTestSuite) executeWorkflow(
 	wfID string, wfFunc interface{}, retValPtr interface{}, args ...interface{},

test/workflow_test.go

@@ -3249,6 +3249,17 @@ func (w *Workflows) SelectorBlockSignal(ctx workflow.Context) (string, error) {
 	return hello, nil
 }
 
+func (w *Workflows) WorkflowClientFromActivity(ctx workflow.Context) error {
+	ctx = workflow.WithActivityOptions(ctx, w.defaultActivityOptions())
+	var activities *Activities
+	var result string
+	err := workflow.ExecuteActivity(ctx, activities.ClientFromActivity).Get(ctx, &result)
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
 func (w *Workflows) register(worker worker.Worker) {
 	worker.RegisterWorkflow(w.ActivityCancelRepro)
 	worker.RegisterWorkflow(w.ActivityCompletionUsingID)
@@ -3386,6 +3397,7 @@ func (w *Workflows) register(worker worker.Worker) {
 	worker.RegisterWorkflow(w.Echo)
 	worker.RegisterWorkflow(w.RunsLocalAndNonlocalActsWithRetries)
 	worker.RegisterWorkflow(w.SelectorBlockSignal)
+	worker.RegisterWorkflow(w.WorkflowClientFromActivity)
 }
 
 func (w *Workflows) defaultActivityOptions() workflow.ActivityOptions {