cloud tasks fanout

tasks.go

	client, err := cloudtasks.NewClient(ctx)
	if err != nil {
		log.Fatalf("NewClient: %v", err)
	}

	destName := fmt.Sprintf("%s", event.Metadata["name"])
	destFormats := event.Metadata["formats"]
	s := strings.Split(fmt.Sprintf("%v", destFormats), ",")

	for _, destFmt := range s {
		data := url.Values{}
		data.Set("format", destFmt)
		data.Set("source", event.Name)
		data.Set("name", destName+"."+destFmt)

		logger.Printf("Converting source [%s], to [%s] as format [%s]", event.Name, destName, destFmt)


		taskID := uuid.New().String()

		parentName := fmt.Sprintf("projects/%s/locations/%s/queues/%s", os.Getenv("GCLOUD_PROJECT"), locationID, queueID)
		queuePath := fmt.Sprintf("%s/tasks/%s", parentName, taskID)
	

		url := cloudRunURL + "/convert"
		req := &taskspb.CreateTaskRequest{
			Parent: parentName,
			Task: &taskspb.Task{
				Name: queuePath,
				MessageType: &taskspb.Task_HttpRequest{
					HttpRequest: &taskspb.HttpRequest{
						HttpMethod: taskspb.HttpMethod_POST,
						Url:        url,
						Headers:    map[string]string{"Content-type": "application/x-www-form-urlencoded"},
						Body:       []byte(data.Encode()),
						AuthorizationHeader: &taskspb.HttpRequest_OidcToken{
							OidcToken: &taskspb.OidcToken{
								ServiceAccountEmail: serviceAccount,
								Audience:            cloudRunURL,
							},
						},
					},
				},
			},
		}
		resp, err := client.CreateTask(ctx, req)
		if err != nil {
			logger.Fatalf("ERROR: %v", err.Error())
		}
		logger.Printf("Enqueued Task %s", resp.GetName())
	}