Method Loop

internal/nodes/http_cache_task_manager.go:95–146

()

}

func (this *HTTPCacheTaskManager) Loop() error {
	rpcClient, err := rpc.SharedRPC()
	if err != nil {
		return err
	}

	resp, err := rpcClient.HTTPCacheTaskKeyRPC.FindDoingHTTPCacheTaskKeys(rpcClient.Context(), &pb.FindDoingHTTPCacheTaskKeysRequest{})
	if err != nil {
		// 忽略连接错误
		if rpc.IsConnError(err) {
			return nil
		}
		return err
	}

	var keys = resp.HttpCacheTaskKeys
	if len(keys) == 0 {
		return nil
	}

	var pbResults = []*pb.UpdateHTTPCacheTaskKeysStatusRequest_KeyResult{}

	var taskGroup = goman.NewTaskGroup()
	for _, key := range keys {
		var taskKey = key
		taskGroup.Run(func() {
			processErr := this.processKey(taskKey)
			var pbResult = &pb.UpdateHTTPCacheTaskKeysStatusRequest_KeyResult{
				Id:            taskKey.Id,
				NodeClusterId: taskKey.NodeClusterId,
				Error:         "",
			}

			if processErr != nil {
				pbResult.Error = processErr.Error()
			}

			taskGroup.Lock()
			pbResults = append(pbResults, pbResult)
			taskGroup.Unlock()
		})
	}

	taskGroup.Wait()

	_, err = rpcClient.HTTPCacheTaskKeyRPC.UpdateHTTPCacheTaskKeysStatus(rpcClient.Context(), &pb.UpdateHTTPCacheTaskKeysStatusRequest{KeyResults: pbResults})
	if err != nil {
		return err
	}

	return nil
}

func (this *HTTPCacheTaskManager) PushTaskKeys(keys []string) {
	select {