Vihren Atlas

matching_engine.go ×9

Code frontier

unlabeled · c_0562b8850138

6 tests · 3678 LOC · 164 files · introduces 0 tests · 55 LOC · 3 files

Introduces — evidence that enters the hierarchy at this concept

Code
11 ranges55 lines · 3 files
Tests
0 tests

Contains — complete concept membership

All code (extent)
663 ranges3678 lines · 164 files · Browse complete extent
All tests (intent)
6 testsBrowse 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.

No tests are introduced at this concept. Its intent tests are introduced by other concepts.

Introduced code

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

3 files ranked by introduced lines: 55 introduced LOC across 11 ranges. Expand a file to inspect source; the > gutter marks introduced lines.

go.temporal.io/server/service/matching/matching_engine.go 49 introduced LOC · 9 ranges

Open complete file

2721 req *matchingservice.PollNexusTaskQueueRequest,
2722 opMetrics metrics.Handler,
2723 > ) (*matchingservice.PollNexusTaskQueueResponse, error) { matching_engine.go
2724 > namespaceID := namespace.ID(req.GetNamespaceId())
2725 > pollerID := req.GetPollerId()
2726 > request := req.Request
2727 > taskQueueName := request.TaskQueue.GetName()
2728 > ns, err := e.namespaceRegistry.GetNamespaceByID(namespaceID)
2729 > if err != nil {
2730 return nil, err
2731 }
2732 > pollLoop: matching_engine.go
2733 > for {
2734 > err := common.IsValidContext(ctx)
2735 > if err != nil {
2736 return nil, err
2737 }
2738 // Add frontend generated pollerID to context so taskqueueMgr can support cancellation of
2739 // long-poll when frontend calls CancelOutstandingPoll API
2740 > pollerCtx := context.WithValue(ctx, pollerIDKey, pollerID) matching_engine.go
2741 > pollerCtx = context.WithValue(pollerCtx, identityKey, request.GetIdentity())
2742 > partition, err := tqid.PartitionFromProto(request.TaskQueue, req.NamespaceId, enumspb.TASK_QUEUE_TYPE_NEXUS)
2743 > if err != nil {
2744 return nil, err
2745 }
2746 > pollMetadata := &pollMetadata{ matching_engine.go
2747 > workerVersionCapabilities: request.WorkerVersionCapabilities,
2748 > deploymentOptions: request.DeploymentOptions,
2749 > forwardedFrom: req.ForwardedSource,
2750 > conditions: req.Conditions,
2751 > workerInstanceKey: request.WorkerInstanceKey,
2752 > }
2753 > task, _, err := e.pollTask(pollerCtx, partition, pollMetadata)
2754 > if err != nil {
2755 if errors.Is(err, errNoTasks) {
2756 return &matchingservice.PollNexusTaskQueueResponse{}, nil
2759 }
2760
2761 > if task.isStarted() { matching_engine.go
2762 // tasks received from remote are already started. So, simply forward the response
2763 return task.pollNexusTaskQueueResponse(), nil
2764 }
2765
2766 > task.finish(taskFinishResult{err: err, consumedToken: true}) matching_engine.go
2767 > if err != nil {
2768 continue pollLoop
2769 }
2770
2771 > taskToken := &tokenspb.NexusTask{ matching_engine.go
2772 > NamespaceId: string(namespaceID),
2773 > TaskQueue: taskQueueName,
2774 > TaskId: task.nexus.taskID,
2775 > TaskQueueKind: partition.Kind(),
2776 > }
2777 > serializedToken, _ := e.tokenSerializer.SerializeNexusTaskToken(taskToken)
2778 >
2779 > nexusReq := task.nexus.request.GetRequest()
2780 > if nexusReq.Header == nil {
2781 nexusReq.Header = make(map[string]string)
2782 }
2783 > nexusReq.Header[nexus.HeaderRequestTimeout] = time.Until(task.nexus.deadline).String() matching_engine.go
2784 > // Java SDK currently expects the header in this form. We should be able to remove this duplication sometime mid 2025.
2785 > nexusReq.Header["Request-Timeout"] = time.Until(task.nexus.deadline).String()
2786 > if !task.nexus.operationDeadline.IsZero() {
2787 nexusReq.Header[nexus.HeaderOperationTimeout] = commonnexus.FormatDuration(time.Until(task.nexus.operationDeadline))
2788 }
2789
2790 > e.emitTaskDispatchLatency(task, partition, req.GetNamespaceId(), ns.Name().String(), pollMetadata) matching_engine.go
2791 > return &matchingservice.PollNexusTaskQueueResponse{
2792 > Response: &workflowservice.PollNexusTaskQueueResponse{
2793 > TaskToken: serializedToken,
2794 > Request: nexusReq,
2795 > PollerScalingDecision: task.pollerScalingDecision,
2796 > },
2797 > }, nil
2798 }
2799 }
go.temporal.io/server/api/matchingservice/v1/request_response.pb.go 4 introduced LOC · 1 range

Open complete file

4183 }
4184
4185 > func (x *PollNexusTaskQueueRequest) GetPollerId() string { request_response.pb.go
4186 > if x != nil {
4187 > return x.PollerId
4188 > }
4189 return ""
4190 }
go.temporal.io/server/service/matching/task.go 2 introduced LOC · 1 range

Open complete file

266 return task.query.createTime
267 } else if task.nexus != nil {
268 > return task.nexus.createTime task.go
269 > }
270
271 return timestamppb.Now()

Direct hierarchy

Parents

Children