Vihren Atlas

workflow_handler.go ×11

Joint frontier

unlabeled · c_d976dc4b51f7

1 test · 45899 LOC · 755 files · introduces 1 test · 117 LOC · 8 files

Introduces — evidence that enters the hierarchy at this concept

Code
30 ranges117 lines · 8 files
Tests
1 test

Contains — complete concept membership

All code (extent)
10865 ranges45899 lines · 755 files · Browse complete extent
All tests (intent)
1 testBrowse complete intent

Neighbourhood graph

The orange circle is the focus. Violet and green circles are every ancestor and descendant, broader and narrower, at any distance; blue squares and pink diamonds are the introduced files and exact introduced tests of every visible concept, not only the focus's. Arrows point from broader to narrower concepts and bridge only concepts omitted from this view. Undirected links show source or test introduction. Concept and file size follows LOC; exact test nodes use test-count units.

Introduced files, introduced tests, and structurally relevant concept specialization

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 native relationship evidence on this page.

Graph controls are ready.

Interactive rendering requires JavaScript and WebGL. Use the native relationship evidence on this page while the interactive map is unavailable.

Native relationship evidence

Every exact file and test below is linked only from the concept that introduces it.

Introduced tests

Every collected test enters the hierarchy at exactly one concept.

1 test introduced at this concept.

Introduced code

Every collected source range enters the hierarchy at exactly one concept.

8 files ranked by introduced lines: 117 introduced LOC across 30 ranges. Expand a file to inspect source; the > gutter marks introduced lines.

go.temporal.io/server/api/matchingservice/v1/request_response.pb.go 28 introduced LOC · 4 ranges

Open complete file

1782 }
1783
1784 > func (x *DescribeTaskQueueRequest) Reset() { request_response.pb.go
1785 > *x = DescribeTaskQueueRequest{}
1786 > mi := &file_temporal_server_api_matchingservice_v1_request_response_proto_msgTypes[19]
1787 > ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
1788 > ms.StoreMessageInfo(mi)
1789 > }
1790
1791 func (x *DescribeTaskQueueRequest) String() string {
1795 func (*DescribeTaskQueueRequest) ProtoMessage() {}
1796
1797 > func (x *DescribeTaskQueueRequest) ProtoReflect() protoreflect.Message { request_response.pb.go
1798 > mi := &file_temporal_server_api_matchingservice_v1_request_response_proto_msgTypes[19]
1799 > if x != nil {
1800 > ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
1801 > if ms.LoadMessageInfo() == nil {
1802 > ms.StoreMessageInfo(mi)
1803 > }
1804 > return ms
1805 }
1806 return mi.MessageOf(x)
1840 }
1841
1842 > func (x *DescribeTaskQueueResponse) Reset() { request_response.pb.go
1843 > *x = DescribeTaskQueueResponse{}
1844 > mi := &file_temporal_server_api_matchingservice_v1_request_response_proto_msgTypes[20]
1845 > ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
1846 > ms.StoreMessageInfo(mi)
1847 > }
1848
1849 func (x *DescribeTaskQueueResponse) String() string {
1853 func (*DescribeTaskQueueResponse) ProtoMessage() {}
1854
1855 > func (x *DescribeTaskQueueResponse) ProtoReflect() protoreflect.Message { request_response.pb.go
1856 > mi := &file_temporal_server_api_matchingservice_v1_request_response_proto_msgTypes[20]
1857 > if x != nil {
1858 > ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
1859 > if ms.LoadMessageInfo() == nil {
1860 > ms.StoreMessageInfo(mi)
1861 > }
1862 > return ms
1863 }
1864 return mi.MessageOf(x)
go.temporal.io/server/service/frontend/workflow_handler.go 28 introduced LOC · 11 ranges

Open complete file

3340 // DescribeTaskQueue returns information about the target taskqueue, right now this API returns the
3341 // pollers which polled this taskqueue in last few minutes.
3342 > func (wh *WorkflowHandler) DescribeTaskQueue(ctx context.Context, request *workflowservice.DescribeTaskQueueRequest) (_ *workflowservice.DescribeTaskQueueResponse, retError error) { workflow_handler.go
3343 > defer log.CapturePanic(wh.logger, &retError)
3344 >
3345 > if request == nil {
3346 return nil, errRequestNotSet
3347 }
3348
3349 > namespaceName := namespace.Name(request.GetNamespace()) workflow_handler.go
3350 > namespaceID, err := wh.namespaceRegistry.GetNamespaceID(namespaceName)
3351 > if err != nil {
3352 return nil, err
3353 }
3354
3355 > if request.ApiMode == enumspb.DESCRIBE_TASK_QUEUE_MODE_UNSPECIFIED { workflow_handler.go
3356 > err = tqid.NormalizeAndValidatePartition(request.TaskQueue, "", wh.config.MaxIDLengthLimit())
3357 > } else {
3358 err = tqid.NormalizeAndValidate(request.TaskQueue, "", wh.config.MaxIDLengthLimit())
3359 }
3360 > if err != nil { workflow_handler.go
3361 return nil, err
3362 }
3363
3364 > if request.TaskQueueType == enumspb.TASK_QUEUE_TYPE_UNSPECIFIED || request.ApiMode == enumspb.DESCRIBE_TASK_QUEUE_MODE_ENHANCED { workflow_handler.go
3365 request.TaskQueueType = enumspb.TASK_QUEUE_TYPE_WORKFLOW
3366 }
3367
3368 > if len(request.TaskQueueTypes) == 0 { workflow_handler.go
3369 > request.TaskQueueTypes = []enumspb.TaskQueueType{enumspb.TASK_QUEUE_TYPE_WORKFLOW, enumspb.TASK_QUEUE_TYPE_ACTIVITY}
3370 > }
3371
3372 > if request.GetReportTaskReachability() && workflow_handler.go
3373 > len(request.GetVersions().GetBuildIds()) > wh.config.ReachabilityQueryBuildIdLimit() {
3374 return nil, serviceerror.NewInvalidArgumentf(
3375 "Too many build ids queried at once with ReportTaskReachability==true, limit: %d", wh.config.ReachabilityQueryBuildIdLimit())
3376 }
3377
3378 > if request.ApiMode == enumspb.DESCRIBE_TASK_QUEUE_MODE_ENHANCED { workflow_handler.go
3379 if request.TaskQueue.Kind == enumspb.TASK_QUEUE_KIND_STICKY {
3380 return nil, errUseEnhancedDescribeOnStickyQueue
3387 }
3388
3389 > matchingResponse, err := wh.matchingClient.DescribeTaskQueue(ctx, &matchingservice.DescribeTaskQueueRequest{ workflow_handler.go
3390 > NamespaceId: namespaceID.String(),
3391 > DescRequest: request,
3392 > })
3393 > if err != nil {
3394 return nil, err
3395 }
3396
3397 > resp := matchingResponse.DescResponse workflow_handler.go
3398 > // Manually parse unknown fields to handle proto incompatibility.
3399 > // TODO: remove this after 1.24.0-m3
3400 > if resp == nil {
3401 resp = &workflowservice.DescribeTaskQueueResponse{}
3402 unknown := []byte(matchingResponse.ProtoReflect().GetUnknown())
go.temporal.io/server/api/matchingservice/v1/service_grpc.pb.go 17 introduced LOC · 5 ranges

Open complete file

380 }
381
382 > func (c *matchingServiceClient) DescribeTaskQueue(ctx context.Context, in *DescribeTaskQueueRequest, opts ...grpc.CallOption) (*DescribeTaskQueueResponse, error) { service_grpc.pb.go
383 > out := new(DescribeTaskQueueResponse)
384 > err := c.cc.Invoke(ctx, MatchingService_DescribeTaskQueue_FullMethodName, in, out, opts...)
385 > if err != nil {
386 return nil, err
387 }
388 > return out, nil service_grpc.pb.go
389 }
390
1209 }
1210
1211 > func _MatchingService_DescribeTaskQueue_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { service_grpc.pb.go
1212 > in := new(DescribeTaskQueueRequest)
1213 > if err := dec(in); err != nil {
1214 return nil, err
1215 }
1216 > if interceptor == nil { service_grpc.pb.go
1217 return srv.(MatchingServiceServer).DescribeTaskQueue(ctx, in)
1218 }
1219 > info := &grpc.UnaryServerInfo{ service_grpc.pb.go
1220 > Server: srv,
1221 > FullMethod: MatchingService_DescribeTaskQueue_FullMethodName,
1222 > }
1223 > handler := func(ctx context.Context, req interface{}) (interface{}, error) {
1224 > return srv.(MatchingServiceServer).DescribeTaskQueue(ctx, req.(*DescribeTaskQueueRequest))
1225 > }
1226 > return interceptor(ctx, in, info, handler)
1227 }
1228
go.temporal.io/server/service/matching/handler.go 17 introduced LOC · 3 ranges

Open complete file

344 ctx context.Context,
345 request *matchingservice.DescribeTaskQueueRequest,
346 > ) (_ *matchingservice.DescribeTaskQueueResponse, retError error) { handler.go
347 > defer log.CapturePanic(h.logger, &retError)
348 > resp, err := h.engine.DescribeTaskQueue(ctx, request)
349 > if err != nil {
350 return nil, err
351 }
352
353 // TODO: remove after 1.24.0-m3
354 > if len(resp.DescResponse.Pollers) > 0 || resp.DescResponse.TaskQueueStatus != nil { handler.go
355 > // Expand pollerinfo and task queue status into tags 1 and 2 for old frontend to handle
356 > // proto incompatibility. This only works without ugly protowire code because
357 > // workflowservice.DescribeTaskQueueResponse and the previous version of
358 > // matchingservice.DescribeTaskQueueResponse have the same first two fields.
359 > oldResp := &workflowservice.DescribeTaskQueueResponse{
360 > Pollers: resp.DescResponse.Pollers,
361 > TaskQueueStatus: resp.DescResponse.TaskQueueStatus,
362 > }
363 > if b, err := proto.Marshal(oldResp); err == nil {
364 > resp.ProtoReflect().SetUnknown(protoreflect.RawFields(b))
365 > }
366 }
367
368 > return resp, nil handler.go
369 }
370
go.temporal.io/server/client/matching/client_gen.go 9 introduced LOC · 3 ranges

Open complete file

195 request *matchingservice.DescribeTaskQueueRequest,
196 opts ...grpc.CallOption,
197 > ) (*matchingservice.DescribeTaskQueueResponse, error) { client_gen.go
198 >
199 > p, err := tqid.PartitionFromProto(request.GetDescRequest().GetTaskQueue(), request.GetNamespaceId(), request.GetDescRequest().GetTaskQueueType())
200 > if err != nil {
201 return nil, err
202 }
203
204 > client, err := c.getClientForTaskQueuePartition(p) client_gen.go
205 > if err != nil {
206 return nil, err
207 }
208 > ctx, cancel := c.createContext(ctx) client_gen.go
209 > defer cancel()
210 > return client.DescribeTaskQueue(ctx, request, opts...)
211 }
212
go.temporal.io/server/client/matching/retryable_client_gen.go 9 introduced LOC · 1 range

Open complete file

181 request *matchingservice.DescribeTaskQueueRequest,
182 opts ...grpc.CallOption,
183 > ) (*matchingservice.DescribeTaskQueueResponse, error) { retryable_client_gen.go
184 > var resp *matchingservice.DescribeTaskQueueResponse
185 > op := func(ctx context.Context) error {
186 > var err error
187 > resp, err = c.client.DescribeTaskQueue(ctx, request, opts...)
188 > return err
189 > }
190 > err := backoff.ThrottleRetryContext(ctx, op, c.policy, c.isRetryable)
191 > return resp, err
192 }
193
go.temporal.io/server/client/matching/metric_client_gen.go 7 introduced LOC · 2 ranges

Open complete file

140 request *matchingservice.DescribeTaskQueueRequest,
141 opts ...grpc.CallOption,
142 > ) (_ *matchingservice.DescribeTaskQueueResponse, retError error) { metric_client_gen.go
143 >
144 > metricsHandler, startTime := c.startMetricsRecording(ctx, "MatchingClientDescribeTaskQueue")
145 > defer func() {
146 > c.finishMetricsRecording(metricsHandler, startTime, retError)
147 > }()
148
149 > return c.client.DescribeTaskQueue(ctx, request, opts...) metric_client_gen.go
150 }
151
go.temporal.io/server/common/rpc/interceptor/routing_key_extractor_gen.go 2 introduced LOC · 1 range

Open complete file

29 case *workflowservice.DescribeScheduleRequest:
30 return namespace.RoutingKey{ID: r.GetScheduleId()}
31 > case *workflowservice.DescribeTaskQueueRequest: routing_key_extractor_gen.go
32 > return namespace.RoutingKey{ID: r.GetTaskQueue().GetName()}
33 case *workflowservice.DescribeWorkerDeploymentRequest:
34 return namespace.RoutingKey{ID: r.GetDeploymentName()}

Direct hierarchy

Children