go.temporal.io/server/client/clientfactory.go
245 LOC · 88 covered · 157 uncovered · 17 ranges · 64 concepts · 3 introducers · 12 tests
File neighbourhood
The centred file is linked to every concept that introduces one of its ranges, every test that runs code from the file, and the gray connector concepts standing between those tests and the file's own introducer concepts. Undirected links join concepts to every file where they introduce source and concepts to the tests they introduce; arrows show specialization between the displayed concepts and bridge only concepts omitted from this view. Concept colors match the source ranges below; connector concepts have no source color and are shown in gray.
Focused file, its introducer and connector concepts, their introduced files, and tests that run code from the file
In the embedded map, ordinary wheel input scrolls the page; use the visible controls to zoom and drag to pan. Open the full-screen map for canvas navigation: wheel pans, Ctrl/Command plus wheel zooms, and arrow keys pan when this region is focused. On touch screens, open the full-screen map to pan or pinch. If JavaScript or WebGL is unavailable, use the related-file, concept, and source links on this page.
Graph controls are ready.
Interactive rendering requires JavaScript and WebGL. Use the related-file, concept, and source links on this page while the interactive map is unavailable.
1
//go:generate mockgen -package $GOPACKAGE -source $GOFILE -destination client_factory_mock.go
2
3
package client
4
5
import (
6
"time"
7
8
"go.temporal.io/api/workflowservice/v1"
9
"go.temporal.io/server/api/adminservice/v1"
10
"go.temporal.io/server/api/historyservice/v1"
11
"go.temporal.io/server/api/matchingservice/v1"
12
"go.temporal.io/server/client/admin"
13
"go.temporal.io/server/client/frontend"
14
"go.temporal.io/server/client/history"
15
"go.temporal.io/server/client/matching"
16
"go.temporal.io/server/common"
17
"go.temporal.io/server/common/dynamicconfig"
18
"go.temporal.io/server/common/log"
19
"go.temporal.io/server/common/membership"
20
"go.temporal.io/server/common/metrics"
21
"go.temporal.io/server/common/namespace"
22
"go.temporal.io/server/common/primitives"
23
"go.temporal.io/server/common/testing/testhooks"
24
"google.golang.org/grpc"
25
)
26
27
type (
28
// Factory can be used to create RPC clients for temporal services
29
Factory interface {
30
NewHistoryClientWithTimeout(timeout time.Duration) (historyservice.HistoryServiceClient, error)
31
NewMatchingClientWithTimeout(namespaceIDToName NamespaceIDToNameFunc, timeout time.Duration, longPollTimeout time.Duration) (matchingservice.MatchingServiceClient, error)
32
NewRemoteFrontendClientWithTimeout(rpcAddress string, timeout time.Duration, longPollTimeout time.Duration) (grpc.ClientConnInterface, workflowservice.WorkflowServiceClient)
33
NewLocalFrontendClientWithTimeout(timeout time.Duration, longPollTimeout time.Duration) (grpc.ClientConnInterface, workflowservice.WorkflowServiceClient, error)
34
NewRemoteAdminClientWithTimeout(rpcAddress string, timeout time.Duration, largeTimeout time.Duration) adminservice.AdminServiceClient
35
NewLocalAdminClientWithTimeout(timeout time.Duration, largeTimeout time.Duration) (adminservice.AdminServiceClient, error)
36
}
37
38
// FactoryProvider can be used to provide a customized client Factory implementation.
39
FactoryProvider interface {
40
NewFactory(
41
rpcFactory common.RPCFactory,
42
monitor membership.Monitor,
43
metricsHandler metrics.Handler,
44
dc *dynamicconfig.Collection,
45
testHooks testhooks.TestHooks,
46
numberOfHistoryShards int32,
47
logger log.Logger,
48
throttledLogger log.Logger,
49
) Factory
50
}
51
52
// NamespaceIDToNameFunc maps a namespaceID to namespace name. Returns error when mapping is not possible.
53
NamespaceIDToNameFunc func(id namespace.ID) (namespace.Name, error)
54
55
rpcClientFactory struct {
56
rpcFactory common.RPCFactory
57
monitor membership.Monitor
58
metricsHandler metrics.Handler
59
dynConfig *dynamicconfig.Collection
60
testHooks testhooks.TestHooks
61
numberOfHistoryShards int32
62
logger log.Logger
63
throttledLogger log.Logger
64
}
65
66
factoryProviderImpl struct {
67
}
68
69
serviceKeyResolverImpl struct {
70
resolver membership.ServiceResolver
71
}
72
)
73
74
// NewFactoryProvider creates a default implementation of FactoryProvider.
76
>
return &factoryProviderImpl{}
77
>
}
78
79
// NewFactory creates an instance of client factory that knows how to dispatch RPC calls.
80
func (p *factoryProviderImpl) NewFactory(
81
rpcFactory common.RPCFactory,
82
monitor membership.Monitor,
83
metricsHandler metrics.Handler,
84
dc *dynamicconfig.Collection,
85
testHooks testhooks.TestHooks,
86
numberOfHistoryShards int32,
87
logger log.Logger,
88
throttledLogger log.Logger,
90
>
return &rpcClientFactory{
91
>
rpcFactory: rpcFactory,
92
>
monitor: monitor,
93
>
metricsHandler: metricsHandler,
94
>
dynConfig: dc,
95
>
testHooks: testHooks,
96
>
numberOfHistoryShards: numberOfHistoryShards,
97
>
logger: logger,
98
>
throttledLogger: throttledLogger,
99
>
}
100
>
}
101
102
>
func (cf *rpcClientFactory) NewHistoryClientWithTimeout(timeout time.Duration) (historyservice.HistoryServiceClient, error) {
fx.go ×44
103
>
resolver, err := cf.monitor.GetResolver(primitives.HistoryService)
104
>
if err != nil {
105
return nil, err
106
}
108
>
cf.dynConfig,
109
>
resolver,
110
>
cf.logger,
111
>
cf.numberOfHistoryShards,
112
>
cf.rpcFactory,
113
>
timeout,
114
>
)
115
>
if cf.metricsHandler != nil {
116
>
client = history.NewMetricClient(client, cf.metricsHandler, cf.logger, cf.throttledLogger)
117
>
}
118
>
return client, nil
119
}
120
121
func (cf *rpcClientFactory) NewMatchingClientWithTimeout(
122
namespaceIDToName NamespaceIDToNameFunc,
123
timeout time.Duration,
124
longPollTimeout time.Duration,
126
>
resolver, err := cf.monitor.GetResolver(primitives.MatchingService)
127
>
if err != nil {
128
return nil, err
129
}
130
132
>
clientProvider := func(clientKey string) (any, func() error, error) {
134
>
return matchingservice.NewMatchingServiceClient(connection), connection.Close, nil
135
>
}
137
>
timeout,
138
>
longPollTimeout,
139
>
common.NewClientCache(keyResolver, clientProvider, cf.logger),
140
>
cf.metricsHandler,
141
>
cf.logger,
142
>
matching.NewLoadBalancer(namespaceIDToName, cf.dynConfig, cf.testHooks),
143
>
dynamicconfig.MatchingSpreadRoutingBatchSize.Get(cf.dynConfig),
144
>
resolver,
145
>
dynamicconfig.MatchingConnectionCloseDelay.Get(cf.dynConfig),
146
>
)
147
>
148
>
if cf.metricsHandler != nil {
149
>
client = matching.NewMetricClient(client, cf.metricsHandler, cf.logger, cf.throttledLogger)
150
>
}
151
>
return client, nil
152
153
}
154
155
func (cf *rpcClientFactory) NewRemoteFrontendClientWithTimeout(
156
rpcAddress string,
157
timeout time.Duration,
158
longPollTimeout time.Duration,
159
) (grpc.ClientConnInterface, workflowservice.WorkflowServiceClient) {
160
connection := cf.rpcFactory.CreateRemoteFrontendGRPCConnection(rpcAddress)
161
client := workflowservice.NewWorkflowServiceClient(connection)
162
return connection, cf.newFrontendClient(client, timeout, longPollTimeout)
163
}
164
165
func (cf *rpcClientFactory) NewLocalFrontendClientWithTimeout(
166
timeout time.Duration,
167
longPollTimeout time.Duration,
169
>
connection := cf.rpcFactory.CreateLocalFrontendGRPCConnection()
170
>
client := workflowservice.NewWorkflowServiceClient(connection)
171
>
return connection, cf.newFrontendClient(client, timeout, longPollTimeout), nil
172
>
}
173
174
func (cf *rpcClientFactory) NewRemoteAdminClientWithTimeout(
175
rpcAddress string,
176
timeout time.Duration,
177
largeTimeout time.Duration,
178
) adminservice.AdminServiceClient {
179
connection := cf.rpcFactory.CreateRemoteFrontendGRPCConnection(rpcAddress)
180
client := adminservice.NewAdminServiceClient(connection)
181
return cf.newAdminClient(client, timeout, largeTimeout)
182
}
183
184
func (cf *rpcClientFactory) NewLocalAdminClientWithTimeout(
185
timeout time.Duration,
186
longPollTimeout time.Duration,
188
>
connection := cf.rpcFactory.CreateLocalFrontendGRPCConnection()
189
>
client := adminservice.NewAdminServiceClient(connection)
190
>
return cf.newAdminClient(client, timeout, longPollTimeout), nil
191
>
}
192
193
func (cf *rpcClientFactory) newAdminClient(
194
client adminservice.AdminServiceClient,
195
timeout time.Duration,
196
longPollTimeout time.Duration,
198
>
client = admin.NewClient(timeout, longPollTimeout, client)
199
>
if cf.metricsHandler != nil {
200
>
client = admin.NewMetricClient(client, cf.metricsHandler, cf.throttledLogger)
201
>
}
202
>
return client
203
}
204
205
func (cf *rpcClientFactory) newFrontendClient(
206
client workflowservice.WorkflowServiceClient,
207
timeout time.Duration,
208
longPollTimeout time.Duration,
210
>
client = frontend.NewClient(timeout, longPollTimeout, client)
211
>
if cf.metricsHandler != nil {
212
>
client = frontend.NewMetricClient(client, cf.metricsHandler, cf.throttledLogger)
213
>
}
214
>
return client
215
}
216
217
>
func newServiceKeyResolver(resolver membership.ServiceResolver) *serviceKeyResolverImpl {
fx.go ×44
218
>
return &serviceKeyResolverImpl{
219
>
resolver: resolver,
220
>
}
221
>
}
222
223
// Lookup returns the address for a node within a batch. key contains the key (including batch
224
// number), and index is the index within the batch. If not using batches, index should be 0.
225
// Note that Lookup(key) and LookupN(key, n)[0] are equal.
227
>
hosts := r.resolver.LookupN(key, index+1)
228
>
if len(hosts) == 0 {
230
>
}
232
index %= len(hosts)
233
}
235
}
236
237
func (r *serviceKeyResolverImpl) GetAllAddresses() ([]string, error) {
238
var all []string
239
240
for _, host := range r.resolver.Members() {
241
all = append(all, host.GetAddress())
242
}
243
244
return all, nil
245
}