Vihren Atlas

activities.go ×10

Joint frontier

unlabeled · c_17e4b77d2deb

1 test · 2682 LOC · 128 files · introduces 1 test · 52 LOC · 2 files

Introduces — evidence that enters the hierarchy at this concept

Code
11 ranges52 lines · 2 files
Tests
1 test

Contains — complete concept membership

All code (extent)
399 ranges2682 lines · 128 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.

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

go.temporal.io/server/service/worker/migration/activities.go 47 introduced LOC · 10 ranges

Open complete file

769 }
770
771 > func (a *activities) SeedReplicationQueueWithUserDataEntries(ctx context.Context, params TaskQueueUserDataReplicationParamsWithNamespace) error { activities.go
772 > if len(params.Namespace) == 0 {
773 return temporal.NewNonRetryableApplicationError("namespace is required", "InvalidArgument", nil)
774 }
775 > if params.PageSize == 0 { activities.go
776 params.PageSize = defaultPageSizeForTaskQueueUserDataReplication
777 }
778 > if params.RPS == 0 { activities.go
779 params.RPS = defaultRPSForTaskQueueUserDataReplication
780 }
781
782 > describeResponse, err := a.frontendClient.DescribeNamespace(ctx, &workflowservice.DescribeNamespaceRequest{ activities.go
783 > Namespace: params.Namespace,
784 > })
785 > if err != nil {
786 return err
787 }
788
789 > rateLimiter := quotas.NewRateLimiter(params.RPS, int(math.Ceil(params.RPS))) activities.go
790 > heartbeatDetails := seedReplicationQueueWithUserDataEntriesHeartbeatDetails{}
791 >
792 > if activity.HasHeartbeatDetails(ctx) {
793 > if err := activity.GetHeartbeatDetails(ctx, &heartbeatDetails); err != nil {
794 return temporal.NewNonRetryableApplicationError("failed to load previous heartbeat details", "TypeError", err)
795 }
796 }
797
798 > for { activities.go
799 > if err := rateLimiter.Wait(ctx); err != nil {
800 return err
801 }
802
803 > request := &persistence.ListTaskQueueUserDataEntriesRequest{ activities.go
804 > NamespaceID: describeResponse.GetNamespaceInfo().Id,
805 > NextPageToken: heartbeatDetails.NextPageToken,
806 > PageSize: params.PageSize,
807 > }
808 > response, err := a.taskManager.ListTaskQueueUserDataEntries(ctx, request)
809 > if err != nil {
810 a.Logger.Error("List task queue user data failed", tag.WorkflowNamespaceID(request.NamespaceID), tag.Error(err))
811 return err
812 }
813 > for idx, entry := range response.Entries { activities.go
814 > if heartbeatDetails.IndexInPage > idx {
815 > continue
816 }
817 > heartbeatDetails.IndexInPage = idx activities.go
818 > activity.RecordHeartbeat(ctx, heartbeatDetails)
819 > err = a.namespaceReplicationQueue.Publish(ctx, &replicationspb.ReplicationTask{
820 > TaskType: enumsspb.REPLICATION_TASK_TYPE_TASK_QUEUE_USER_DATA,
821 > Attributes: &replicationspb.ReplicationTask_TaskQueueUserDataAttributes{
822 > TaskQueueUserDataAttributes: &replicationspb.TaskQueueUserDataAttributes{
823 > NamespaceId: request.NamespaceID,
824 > TaskQueueName: entry.TaskQueue,
825 > UserData: entry.UserData.GetData(),
826 > },
827 > },
828 > })
829 > if err != nil {
830 > a.Logger.Error("Inserting into namespace replication queue failed", tag.WorkflowNamespaceID(request.NamespaceID), tag.Error(err))
831 > return err
832 > }
833 }
834 > if len(response.NextPageToken) == 0 { activities.go
835 > return nil
836 > }
837 > heartbeatDetails.NextPageToken = response.NextPageToken
838 > heartbeatDetails.IndexInPage = 0
839 > activity.RecordHeartbeat(ctx, heartbeatDetails)
840 }
841 }
go.temporal.io/server/api/replication/v1/message.pb.go 5 introduced LOC · 1 range

Open complete file

157 }
158
159 > func (x *ReplicationTask) GetTaskQueueUserDataAttributes() *TaskQueueUserDataAttributes { message.pb.go
160 > if x != nil {
161 > if x, ok := x.Attributes.(*ReplicationTask_TaskQueueUserDataAttributes); ok {
162 > return x.TaskQueueUserDataAttributes
163 > }
164 }
165 return nil

Direct hierarchy

Children