tasks/chasm/lib/scheduler/gen/schedulerpb/v1/service_client.pb.go

// Code generated by protoc-gen-go-chasm. DO NOT EDIT.
package schedulerpb

import (
	"context"
	"time"

	"go.temporal.io/server/client/history"
	"go.temporal.io/server/common"
	"go.temporal.io/server/common/backoff"
	"go.temporal.io/server/common/config"
	"go.temporal.io/server/common/dynamicconfig"
	"go.temporal.io/server/common/headers"
	"go.temporal.io/server/common/log"
	"go.temporal.io/server/common/membership"
	"go.temporal.io/server/common/metrics"
	"go.temporal.io/server/common/primitives"
	"google.golang.org/grpc"
)

// SchedulerServiceLayeredClient is a client for SchedulerService.
type SchedulerServiceLayeredClient struct {
	metricsHandler metrics.Handler
	numShards      int32
	redirector     history.Redirector[SchedulerServiceClient]
	retryPolicy    backoff.RetryPolicy
}

// NewSchedulerServiceLayeredClient initializes a new SchedulerServiceLayeredClient.
func NewSchedulerServiceLayeredClient(
	dc *dynamicconfig.Collection,
	rpcFactory common.RPCFactory,
	monitor membership.Monitor,
	config *config.Persistence,
	logger log.Logger,
	metricsHandler metrics.Handler,
) (SchedulerServiceClient, error) {
	resolver, err := monitor.GetResolver(primitives.HistoryService)
	if err != nil {
		return nil, err
	}
	connections := history.NewConnectionPool(resolver, rpcFactory, NewSchedulerServiceClient)
	var redirector history.Redirector[SchedulerServiceClient]
	if dynamicconfig.HistoryClientOwnershipCachingEnabled.Get(dc)() {
		redirector = history.NewCachingRedirector(
			connections,
			resolver,
			logger,
			dynamicconfig.HistoryClientOwnershipCachingStaleTTL.Get(dc),
		)
	} else {
		redirector = history.NewBasicRedirector(connections, resolver)
	}
	return &SchedulerServiceLayeredClient{
		metricsHandler: metricsHandler,
		redirector:     redirector,
		numShards:      config.NumHistoryShards,
		retryPolicy:    common.CreateHistoryClientRetryPolicy(),
	}, nil
}
func (c *SchedulerServiceLayeredClient) callCreateScheduleNoRetry(
	ctx context.Context,
	request *CreateScheduleRequest,
	opts ...grpc.CallOption,
) (*CreateScheduleResponse, error) {
	var response *CreateScheduleResponse
	var err error
	startTime := time.Now().UTC()
	// the caller is a namespace, hence the tag below.
	caller := headers.GetCallerInfo(ctx).CallerName
	metricsHandler := c.metricsHandler.WithTags(
		metrics.OperationTag("SchedulerService.CreateSchedule"),
		metrics.NamespaceTag(caller),
		metrics.ServiceRoleTag(metrics.HistoryRoleTagValue),
	)
	metrics.ClientRequests.With(metricsHandler).Record(1)
	defer func() {
		if err != nil {
			metrics.ClientFailures.With(metricsHandler).Record(1, metrics.ServiceErrorTypeTag(err))
		}
		metrics.ClientLatency.With(metricsHandler).Record(time.Since(startTime))
	}()
	shardID := common.WorkflowIDToHistoryShard(request.GetNamespaceId(), request.GetFrontendRequest().GetScheduleId(), c.numShards)
	op := func(ctx context.Context, client SchedulerServiceClient) error {
		var err error
		ctx, cancel := context.WithTimeout(ctx, history.DefaultTimeout)
		defer cancel()
		response, err = client.CreateSchedule(ctx, request, opts...)
		return err
	}
	err = c.redirector.Execute(ctx, shardID, op)
	return response, err
}
func (c *SchedulerServiceLayeredClient) CreateSchedule(
	ctx context.Context,
	request *CreateScheduleRequest,
	opts ...grpc.CallOption,
) (*CreateScheduleResponse, error) {
	call := func(ctx context.Context) (*CreateScheduleResponse, error) {
		return c.callCreateScheduleNoRetry(ctx, request, opts...)
	}
	return backoff.ThrottleRetryContextWithReturn(ctx, call, c.retryPolicy, common.IsServiceClientTransientError)
}
func (c *SchedulerServiceLayeredClient) callUpdateScheduleNoRetry(
	ctx context.Context,
	request *UpdateScheduleRequest,
	opts ...grpc.CallOption,
) (*UpdateScheduleResponse, error) {
	var response *UpdateScheduleResponse
	var err error
	startTime := time.Now().UTC()
	// the caller is a namespace, hence the tag below.
	caller := headers.GetCallerInfo(ctx).CallerName
	metricsHandler := c.metricsHandler.WithTags(
		metrics.OperationTag("SchedulerService.UpdateSchedule"),
		metrics.NamespaceTag(caller),
		metrics.ServiceRoleTag(metrics.HistoryRoleTagValue),
	)
	metrics.ClientRequests.With(metricsHandler).Record(1)
	defer func() {
		if err != nil {
			metrics.ClientFailures.With(metricsHandler).Record(1, metrics.ServiceErrorTypeTag(err))
		}
		metrics.ClientLatency.With(metricsHandler).Record(time.Since(startTime))
	}()
	shardID := common.WorkflowIDToHistoryShard(request.GetNamespaceId(), request.GetFrontendRequest().GetScheduleId(), c.numShards)
	op := func(ctx context.Context, client SchedulerServiceClient) error {
		var err error
		ctx, cancel := context.WithTimeout(ctx, history.DefaultTimeout)
		defer cancel()
		response, err = client.UpdateSchedule(ctx, request, opts...)
		return err
	}
	err = c.redirector.Execute(ctx, shardID, op)
	return response, err
}
func (c *SchedulerServiceLayeredClient) UpdateSchedule(
	ctx context.Context,
	request *UpdateScheduleRequest,
	opts ...grpc.CallOption,
) (*UpdateScheduleResponse, error) {
	call := func(ctx context.Context) (*UpdateScheduleResponse, error) {
		return c.callUpdateScheduleNoRetry(ctx, request, opts...)
	}
	return backoff.ThrottleRetryContextWithReturn(ctx, call, c.retryPolicy, common.IsServiceClientTransientError)
}
func (c *SchedulerServiceLayeredClient) callPatchScheduleNoRetry(
	ctx context.Context,
	request *PatchScheduleRequest,
	opts ...grpc.CallOption,
) (*PatchScheduleResponse, error) {
	var response *PatchScheduleResponse
	var err error
	startTime := time.Now().UTC()
	// the caller is a namespace, hence the tag below.
	caller := headers.GetCallerInfo(ctx).CallerName
	metricsHandler := c.metricsHandler.WithTags(
		metrics.OperationTag("SchedulerService.PatchSchedule"),
		metrics.NamespaceTag(caller),
		metrics.ServiceRoleTag(metrics.HistoryRoleTagValue),
	)
	metrics.ClientRequests.With(metricsHandler).Record(1)
	defer func() {
		if err != nil {
			metrics.ClientFailures.With(metricsHandler).Record(1, metrics.ServiceErrorTypeTag(err))
		}
		metrics.ClientLatency.With(metricsHandler).Record(time.Since(startTime))
	}()
	shardID := common.WorkflowIDToHistoryShard(request.GetNamespaceId(), request.GetFrontendRequest().GetScheduleId(), c.numShards)
	op := func(ctx context.Context, client SchedulerServiceClient) error {
		var err error
		ctx, cancel := context.WithTimeout(ctx, history.DefaultTimeout)
		defer cancel()
		response, err = client.PatchSchedule(ctx, request, opts...)
		return err
	}
	err = c.redirector.Execute(ctx, shardID, op)
	return response, err
}
func (c *SchedulerServiceLayeredClient) PatchSchedule(
	ctx context.Context,
	request *PatchScheduleRequest,
	opts ...grpc.CallOption,
) (*PatchScheduleResponse, error) {
	call := func(ctx context.Context) (*PatchScheduleResponse, error) {
		return c.callPatchScheduleNoRetry(ctx, request, opts...)
	}
	return backoff.ThrottleRetryContextWithReturn(ctx, call, c.retryPolicy, common.IsServiceClientTransientError)
}
func (c *SchedulerServiceLayeredClient) callDeleteScheduleNoRetry(
	ctx context.Context,
	request *DeleteScheduleRequest,
	opts ...grpc.CallOption,
) (*DeleteScheduleResponse, error) {
	var response *DeleteScheduleResponse
	var err error
	startTime := time.Now().UTC()
	// the caller is a namespace, hence the tag below.
	caller := headers.GetCallerInfo(ctx).CallerName
	metricsHandler := c.metricsHandler.WithTags(
		metrics.OperationTag("SchedulerService.DeleteSchedule"),
		metrics.NamespaceTag(caller),
		metrics.ServiceRoleTag(metrics.HistoryRoleTagValue),
	)
	metrics.ClientRequests.With(metricsHandler).Record(1)
	defer func() {
		if err != nil {
			metrics.ClientFailures.With(metricsHandler).Record(1, metrics.ServiceErrorTypeTag(err))
		}
		metrics.ClientLatency.With(metricsHandler).Record(time.Since(startTime))
	}()
	shardID := common.WorkflowIDToHistoryShard(request.GetNamespaceId(), request.GetFrontendRequest().GetScheduleId(), c.numShards)
	op := func(ctx context.Context, client SchedulerServiceClient) error {
		var err error
		ctx, cancel := context.WithTimeout(ctx, history.DefaultTimeout)
		defer cancel()
		response, err = client.DeleteSchedule(ctx, request, opts...)
		return err
	}
	err = c.redirector.Execute(ctx, shardID, op)
	return response, err
}
func (c *SchedulerServiceLayeredClient) DeleteSchedule(
	ctx context.Context,
	request *DeleteScheduleRequest,
	opts ...grpc.CallOption,
) (*DeleteScheduleResponse, error) {
	call := func(ctx context.Context) (*DeleteScheduleResponse, error) {
		return c.callDeleteScheduleNoRetry(ctx, request, opts...)
	}
	return backoff.ThrottleRetryContextWithReturn(ctx, call, c.retryPolicy, common.IsServiceClientTransientError)
}
func (c *SchedulerServiceLayeredClient) callDescribeScheduleNoRetry(
	ctx context.Context,
	request *DescribeScheduleRequest,
	opts ...grpc.CallOption,
) (*DescribeScheduleResponse, error) {
	var response *DescribeScheduleResponse
	var err error
	startTime := time.Now().UTC()
	// the caller is a namespace, hence the tag below.
	caller := headers.GetCallerInfo(ctx).CallerName
	metricsHandler := c.metricsHandler.WithTags(
		metrics.OperationTag("SchedulerService.DescribeSchedule"),
		metrics.NamespaceTag(caller),
		metrics.ServiceRoleTag(metrics.HistoryRoleTagValue),
	)
	metrics.ClientRequests.With(metricsHandler).Record(1)
	defer func() {
		if err != nil {
			metrics.ClientFailures.With(metricsHandler).Record(1, metrics.ServiceErrorTypeTag(err))
		}
		metrics.ClientLatency.With(metricsHandler).Record(time.Since(startTime))
	}()
	shardID := common.WorkflowIDToHistoryShard(request.GetNamespaceId(), request.GetFrontendRequest().GetScheduleId(), c.numShards)
	op := func(ctx context.Context, client SchedulerServiceClient) error {
		var err error
		ctx, cancel := context.WithTimeout(ctx, history.DefaultTimeout)
		defer cancel()
		response, err = client.DescribeSchedule(ctx, request, opts...)
		return err
	}
	err = c.redirector.Execute(ctx, shardID, op)
	return response, err
}
func (c *SchedulerServiceLayeredClient) DescribeSchedule(
	ctx context.Context,
	request *DescribeScheduleRequest,
	opts ...grpc.CallOption,
) (*DescribeScheduleResponse, error) {
	call := func(ctx context.Context) (*DescribeScheduleResponse, error) {
		return c.callDescribeScheduleNoRetry(ctx, request, opts...)
	}
	return backoff.ThrottleRetryContextWithReturn(ctx, call, c.retryPolicy, common.IsServiceClientTransientError)
}
func (c *SchedulerServiceLayeredClient) callListScheduleMatchingTimesNoRetry(
	ctx context.Context,
	request *ListScheduleMatchingTimesRequest,
	opts ...grpc.CallOption,
) (*ListScheduleMatchingTimesResponse, error) {
	var response *ListScheduleMatchingTimesResponse
	var err error
	startTime := time.Now().UTC()
	// the caller is a namespace, hence the tag below.
	caller := headers.GetCallerInfo(ctx).CallerName
	metricsHandler := c.metricsHandler.WithTags(
		metrics.OperationTag("SchedulerService.ListScheduleMatchingTimes"),
		metrics.NamespaceTag(caller),
		metrics.ServiceRoleTag(metrics.HistoryRoleTagValue),
	)
	metrics.ClientRequests.With(metricsHandler).Record(1)
	defer func() {
		if err != nil {
			metrics.ClientFailures.With(metricsHandler).Record(1, metrics.ServiceErrorTypeTag(err))
		}
		metrics.ClientLatency.With(metricsHandler).Record(time.Since(startTime))
	}()
	shardID := common.WorkflowIDToHistoryShard(request.GetNamespaceId(), request.GetFrontendRequest().GetScheduleId(), c.numShards)
	op := func(ctx context.Context, client SchedulerServiceClient) error {
		var err error
		ctx, cancel := context.WithTimeout(ctx, history.DefaultTimeout)
		defer cancel()
		response, err = client.ListScheduleMatchingTimes(ctx, request, opts...)
		return err
	}
	err = c.redirector.Execute(ctx, shardID, op)
	return response, err
}
func (c *SchedulerServiceLayeredClient) ListScheduleMatchingTimes(
	ctx context.Context,
	request *ListScheduleMatchingTimesRequest,
	opts ...grpc.CallOption,
) (*ListScheduleMatchingTimesResponse, error) {
	call := func(ctx context.Context) (*ListScheduleMatchingTimesResponse, error) {
		return c.callListScheduleMatchingTimesNoRetry(ctx, request, opts...)
	}
	return backoff.ThrottleRetryContextWithReturn(ctx, call, c.retryPolicy, common.IsServiceClientTransientError)
}
func (c *SchedulerServiceLayeredClient) callCreateFromMigrationStateNoRetry(
	ctx context.Context,
	request *CreateFromMigrationStateRequest,
	opts ...grpc.CallOption,
) (*CreateFromMigrationStateResponse, error) {
	var response *CreateFromMigrationStateResponse
	var err error
	startTime := time.Now().UTC()
	// the caller is a namespace, hence the tag below.
	caller := headers.GetCallerInfo(ctx).CallerName
	metricsHandler := c.metricsHandler.WithTags(
		metrics.OperationTag("SchedulerService.CreateFromMigrationState"),
		metrics.NamespaceTag(caller),
		metrics.ServiceRoleTag(metrics.HistoryRoleTagValue),
	)
	metrics.ClientRequests.With(metricsHandler).Record(1)
	defer func() {
		if err != nil {
			metrics.ClientFailures.With(metricsHandler).Record(1, metrics.ServiceErrorTypeTag(err))
		}
		metrics.ClientLatency.With(metricsHandler).Record(time.Since(startTime))
	}()
	shardID := common.WorkflowIDToHistoryShard(request.GetNamespaceId(), request.GetState().GetSchedulerState().GetScheduleId(), c.numShards)
	op := func(ctx context.Context, client SchedulerServiceClient) error {
		var err error
		ctx, cancel := context.WithTimeout(ctx, history.DefaultTimeout)
		defer cancel()
		response, err = client.CreateFromMigrationState(ctx, request, opts...)
		return err
	}
	err = c.redirector.Execute(ctx, shardID, op)
	return response, err
}
func (c *SchedulerServiceLayeredClient) CreateFromMigrationState(
	ctx context.Context,
	request *CreateFromMigrationStateRequest,
	opts ...grpc.CallOption,
) (*CreateFromMigrationStateResponse, error) {
	call := func(ctx context.Context) (*CreateFromMigrationStateResponse, error) {
		return c.callCreateFromMigrationStateNoRetry(ctx, request, opts...)
	}
	return backoff.ThrottleRetryContextWithReturn(ctx, call, c.retryPolicy, common.IsServiceClientTransientError)
}
func (c *SchedulerServiceLayeredClient) callCreateSentinelNoRetry(
	ctx context.Context,
	request *CreateSentinelRequest,
	opts ...grpc.CallOption,
) (*CreateSentinelResponse, error) {
	var response *CreateSentinelResponse
	var err error
	startTime := time.Now().UTC()
	// the caller is a namespace, hence the tag below.
	caller := headers.GetCallerInfo(ctx).CallerName
	metricsHandler := c.metricsHandler.WithTags(
		metrics.OperationTag("SchedulerService.CreateSentinel"),
		metrics.NamespaceTag(caller),
		metrics.ServiceRoleTag(metrics.HistoryRoleTagValue),
	)
	metrics.ClientRequests.With(metricsHandler).Record(1)
	defer func() {
		if err != nil {
			metrics.ClientFailures.With(metricsHandler).Record(1, metrics.ServiceErrorTypeTag(err))
		}
		metrics.ClientLatency.With(metricsHandler).Record(time.Since(startTime))
	}()
	shardID := common.WorkflowIDToHistoryShard(request.GetNamespaceId(), request.GetScheduleId(), c.numShards)
	op := func(ctx context.Context, client SchedulerServiceClient) error {
		var err error
		ctx, cancel := context.WithTimeout(ctx, history.DefaultTimeout)
		defer cancel()
		response, err = client.CreateSentinel(ctx, request, opts...)
		return err
	}
	err = c.redirector.Execute(ctx, shardID, op)
	return response, err
}
func (c *SchedulerServiceLayeredClient) CreateSentinel(
	ctx context.Context,
	request *CreateSentinelRequest,
	opts ...grpc.CallOption,
) (*CreateSentinelResponse, error) {
	call := func(ctx context.Context) (*CreateSentinelResponse, error) {
		return c.callCreateSentinelNoRetry(ctx, request, opts...)
	}
	return backoff.ThrottleRetryContextWithReturn(ctx, call, c.retryPolicy, common.IsServiceClientTransientError)
}
func (c *SchedulerServiceLayeredClient) callMigrateToWorkflowNoRetry(
	ctx context.Context,
	request *MigrateToWorkflowRequest,
	opts ...grpc.CallOption,
) (*MigrateToWorkflowResponse, error) {
	var response *MigrateToWorkflowResponse
	var err error
	startTime := time.Now().UTC()
	// the caller is a namespace, hence the tag below.
	caller := headers.GetCallerInfo(ctx).CallerName
	metricsHandler := c.metricsHandler.WithTags(
		metrics.OperationTag("SchedulerService.MigrateToWorkflow"),
		metrics.NamespaceTag(caller),
		metrics.ServiceRoleTag(metrics.HistoryRoleTagValue),
	)
	metrics.ClientRequests.With(metricsHandler).Record(1)
	defer func() {
		if err != nil {
			metrics.ClientFailures.With(metricsHandler).Record(1, metrics.ServiceErrorTypeTag(err))
		}
		metrics.ClientLatency.With(metricsHandler).Record(time.Since(startTime))
	}()
	shardID := common.WorkflowIDToHistoryShard(request.GetNamespaceId(), request.GetScheduleId(), c.numShards)
	op := func(ctx context.Context, client SchedulerServiceClient) error {
		var err error
		ctx, cancel := context.WithTimeout(ctx, history.DefaultTimeout)
		defer cancel()
		response, err = client.MigrateToWorkflow(ctx, request, opts...)
		return err
	}
	err = c.redirector.Execute(ctx, shardID, op)
	return response, err
}
func (c *SchedulerServiceLayeredClient) MigrateToWorkflow(
	ctx context.Context,
	request *MigrateToWorkflowRequest,
	opts ...grpc.CallOption,
) (*MigrateToWorkflowResponse, error) {
	call := func(ctx context.Context) (*MigrateToWorkflowResponse, error) {
		return c.callMigrateToWorkflowNoRetry(ctx, request, opts...)
	}
	return backoff.ThrottleRetryContextWithReturn(ctx, call, c.retryPolicy, common.IsServiceClientTransientError)
}