go.temporal.io/server/client/client_bean.go
248 LOC · 79 covered · 169 uncovered · 14 ranges · 64 concepts · 2 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_bean_mock.go
2
3
package client
4
5
import (
6
"fmt"
7
"sync"
8
"sync/atomic"
9
10
"go.temporal.io/api/serviceerror"
11
"go.temporal.io/api/workflowservice/v1"
12
"go.temporal.io/server/api/adminservice/v1"
13
"go.temporal.io/server/api/historyservice/v1"
14
"go.temporal.io/server/api/matchingservice/v1"
15
"go.temporal.io/server/client/admin"
16
"go.temporal.io/server/client/frontend"
17
"go.temporal.io/server/client/history"
18
"go.temporal.io/server/client/matching"
19
"go.temporal.io/server/common/cluster"
20
"google.golang.org/grpc"
21
)
22
23
type (
24
// Bean is a collection of clients
25
Bean interface {
26
GetHistoryClient() historyservice.HistoryServiceClient
27
GetMatchingClient(namespaceIDToName NamespaceIDToNameFunc) (matchingservice.MatchingServiceClient, error)
28
GetFrontendClient() workflowservice.WorkflowServiceClient
29
GetRemoteAdminClient(string) (adminservice.AdminServiceClient, error)
30
GetRemoteFrontendClient(string) (grpc.ClientConnInterface, workflowservice.WorkflowServiceClient, error)
31
// Close deterministically releases bean-held resources on shutdown: it
32
// stops the history and matching clients (their daemon goroutines and
33
// cached gRPC connections) and unregisters the cluster metadata change
34
// callback.
35
Close()
36
}
37
38
frontendClient struct {
39
connection grpc.ClientConnInterface
40
workflowservice.WorkflowServiceClient
41
}
42
43
clientBeanImpl struct {
44
sync.Mutex
45
historyClient historyservice.HistoryServiceClient
46
matchingClient atomic.Value
47
clusterMetadata cluster.Metadata
48
factory Factory
49
50
adminClientsLock sync.RWMutex
51
adminClients map[string]adminservice.AdminServiceClient
52
frontendClientsLock sync.RWMutex
53
frontendClients map[string]frontendClient
54
}
55
)
56
57
// NewClientBean provides a collection of clients
58
>
func NewClientBean(factory Factory, clusterMetadata cluster.Metadata) (Bean, error) {
fx.go ×44
59
>
60
>
historyClient, err := factory.NewHistoryClientWithTimeout(history.DefaultTimeout)
61
>
if err != nil {
62
return nil, err
63
}
64
66
>
frontendClients := map[string]frontendClient{}
67
>
68
>
currentClusterName := clusterMetadata.GetCurrentClusterName()
69
>
// Init local cluster client with membership info
70
>
adminClient, err := factory.NewLocalAdminClientWithTimeout(
71
>
admin.DefaultTimeout,
72
>
admin.DefaultLargeTimeout,
73
>
)
74
>
if err != nil {
75
return nil, err
76
}
78
>
frontend.DefaultTimeout,
79
>
frontend.DefaultLongPollTimeout,
80
>
)
81
>
if err != nil {
82
return nil, err
83
}
85
>
frontendClients[currentClusterName] = frontendClient{
86
>
connection: conn,
87
>
WorkflowServiceClient: client,
88
>
}
89
>
90
>
bean := &clientBeanImpl{
91
>
factory: factory,
92
>
historyClient: historyClient,
93
>
clusterMetadata: clusterMetadata,
94
>
adminClients: adminClients,
95
>
frontendClients: frontendClients,
96
>
}
97
>
bean.registerClientEviction()
98
>
return bean, nil
99
}
100
102
>
currentCluster := h.clusterMetadata.GetCurrentClusterName()
103
>
h.clusterMetadata.RegisterMetadataChangeCallback(
104
>
h,
105
>
func(oldClusterMetadata map[string]*cluster.ClusterInformation, newClusterMetadata map[string]*cluster.ClusterInformation) {
106
>
for clusterName := range newClusterMetadata {
107
>
if clusterName == currentCluster {
108
>
continue
109
}
110
h.adminClientsLock.Lock()
111
delete(h.adminClients, clusterName)
112
h.adminClientsLock.Unlock()
113
h.frontendClientsLock.Lock()
114
delete(h.frontendClients, clusterName)
115
h.frontendClientsLock.Unlock()
116
}
117
})
118
}
119
120
// Close releases the resources held by the bean's clients. See the Bean
121
// interface for details. It is safe to call more than once.
123
>
h.clusterMetadata.UnRegisterMetadataChangeCallback(h)
124
>
125
>
// The history and matching client wrapper chains implement Stop();
126
>
// stopping them releases their daemon goroutines and cached gRPC
127
>
// connections.
128
>
if s, ok := h.historyClient.(interface{ Stop() }); ok {
129
>
s.Stop()
130
>
}
131
>
if mc := h.matchingClient.Load(); mc != nil {
132
>
if s, ok := mc.(interface{ Stop() }); ok {
133
>
s.Stop()
134
>
}
135
}
136
}
137
139
>
return h.historyClient
140
>
}
141
142
>
func (h *clientBeanImpl) GetMatchingClient(namespaceIDToName NamespaceIDToNameFunc) (matchingservice.MatchingServiceClient, error) {
fx.go ×44
143
>
if client := h.matchingClient.Load(); client != nil {
144
return client.(matchingservice.MatchingServiceClient), nil
145
}
147
}
148
149
>
func (h *clientBeanImpl) GetFrontendClient() workflowservice.WorkflowServiceClient {
fx.go ×44
150
>
return h.frontendClients[h.clusterMetadata.GetCurrentClusterName()]
151
>
}
152
153
>
func (h *clientBeanImpl) GetRemoteAdminClient(cluster string) (adminservice.AdminServiceClient, error) {
fx.go ×44
154
>
h.adminClientsLock.RLock()
155
>
client, ok := h.adminClients[cluster]
156
>
h.adminClientsLock.RUnlock()
157
>
if ok {
158
>
return client, nil
159
>
}
160
161
clusterInfo, clusterFound := h.clusterMetadata.GetAllClusterInfo()[cluster]
162
if !clusterFound {
163
// We intentionally return internal error here.
164
// This error could only happen with internal mis-configuration.
165
// This can happen when a namespace is config for multiple clusters. But those clusters are not connected.
166
// We also have logic in task processing to drop tasks when namespace cluster exclude a local cluster.
167
return nil, &serviceerror.Internal{
168
Message: fmt.Sprintf(
169
"Unknown cluster name: %v with given cluster information map: %v.",
170
cluster,
171
clusterInfo,
172
),
173
}
174
}
175
176
h.adminClientsLock.Lock()
177
defer h.adminClientsLock.Unlock()
178
client, ok = h.adminClients[cluster]
179
if ok {
180
return client, nil
181
}
182
183
client = h.factory.NewRemoteAdminClientWithTimeout(
184
clusterInfo.RPCAddress,
185
admin.DefaultTimeout,
186
admin.DefaultLargeTimeout,
187
)
188
h.adminClients[cluster] = client
189
return client, nil
190
}
191
192
func (h *clientBeanImpl) GetRemoteFrontendClient(clusterName string) (grpc.ClientConnInterface, workflowservice.WorkflowServiceClient, error) {
193
h.frontendClientsLock.RLock()
194
client, ok := h.frontendClients[clusterName]
195
h.frontendClientsLock.RUnlock()
196
if ok {
197
return client.connection, client, nil
198
}
199
200
clusterInfo, clusterFound := h.clusterMetadata.GetAllClusterInfo()[clusterName]
201
if !clusterFound {
202
// We intentionally return internal error here.
203
// This error could only happen with internal mis-configuration.
204
// This can happen when a namespace is config for multiple clusters. But those clusters are not connected.
205
// We also have logic in task processing to drop tasks when namespace cluster exclude a local cluster.
206
return nil, nil, &serviceerror.Internal{
207
Message: fmt.Sprintf(
208
"Unknown clusterName name: %v with given clusterName information map: %v.",
209
clusterName,
210
clusterInfo,
211
),
212
}
213
}
214
215
h.frontendClientsLock.Lock()
216
defer h.frontendClientsLock.Unlock()
217
218
client, ok = h.frontendClients[clusterName]
219
if ok {
220
return client.connection, client, nil
221
}
222
223
conn, fClient := h.factory.NewRemoteFrontendClientWithTimeout(
224
clusterInfo.RPCAddress,
225
frontend.DefaultTimeout,
226
frontend.DefaultLongPollTimeout,
227
)
228
client = frontendClient{
229
connection: conn,
230
WorkflowServiceClient: fClient,
231
}
232
h.frontendClients[clusterName] = client
233
return client.connection, client, nil
234
}
235
236
>
func (h *clientBeanImpl) lazyInitMatchingClient(namespaceIDToName NamespaceIDToNameFunc) (matchingservice.MatchingServiceClient, error) {
fx.go ×44
237
>
h.Lock()
238
>
defer h.Unlock()
239
>
if cached := h.matchingClient.Load(); cached != nil {
240
return cached.(matchingservice.MatchingServiceClient), nil
241
}
242
>
client, err := h.factory.NewMatchingClientWithTimeout(namespaceIDToName, matching.DefaultTimeout, matching.DefaultLongPollTimeout)
fx.go ×44
243
>
if err != nil {
244
return nil, err
245
}
247
>
return client, nil
248
}