2641
}
2642
2643
>
func (e *matchingEngineImpl) DispatchNexusTask(ctx context.Context, request *matchingservice.DispatchNexusTaskRequest) (*matchingservice.DispatchNexusTaskResponse, error) {
matching_engine.go
2644
>
partition, err := tqid.PartitionFromProto(request.GetTaskQueue(), request.GetNamespaceId(), enumspb.TASK_QUEUE_TYPE_NEXUS)
2645
>
if err != nil {
2646
return nil, err
2647
}
2648
>
pm, _, err := e.getTaskQueuePartitionManager(ctx, partition, true, loadCauseNexusTask)
matching_engine.go
2649
>
if err != nil {
2650
return nil, err
2651
}
2652
2654
>
2655
>
namespaceID := namespace.ID(request.GetNamespaceId())
2656
>
ns, err := e.namespaceRegistry.GetNamespaceByID(namespaceID)
2657
>
if err != nil {
2658
return nil, err
2659
}
2660
2661
// Buffer the deadline so we can still respond with timeout if we hit the deadline while dispatching
2662
>
ctx, cancel := contextutil.WithDeadlineBuffer(ctx, matching.DefaultTimeout, e.config.MinDispatchTaskTimeout(ns.Name().String()))
matching_engine.go
2663
>
defer cancel()
2664
>
2665
>
// First allocate a result channel and register it so that when the task is completed locally (without forwarding) the
2666
>
// result can be sent on this channel.
2667
>
resultCh := make(chan *nexusResult, 1)
2668
>
e.nexusResults.Set(taskID, resultCh)
2669
>
defer e.nexusResults.Delete(taskID)
2670
>
2671
>
resp, err := pm.DispatchNexusTask(ctx, taskID, request)
2672
>
2673
>
if err != nil {
2674
if ctx.Err() != nil {
2675
// The context deadline has expired if it reaches here; return an explicit timeout response to the caller.