func ExportConcurrent()

in internal/client/integrations/integrations.go [1049:1113]
53 lines of code
10 McCabe index (conditional complexity)

func ExportConcurrent(folder string, numConnections int) error {
	// Set export settings
	apiclient.SetExportToFile(folder)
	apiclient.ClientPrintHttpResponse.Set(false)
	defer apiclient.ClientPrintHttpResponse.Set(apiclient.GetCmdPrintHttpResponseSetting())

	pageToken := ""
	lintegrations := listintegrations{}

	for {
		l := listintegrations{}
		listRespBytes, err := List(maxPageSize, pageToken, "", "")
		if err != nil {
			return fmt.Errorf("failed to fetch Integrations: %w", err)
		}
		err = json.Unmarshal(listRespBytes, &l)
		if err != nil {
			return fmt.Errorf("failed to unmarshall: %w", err)
		}
		lintegrations.Integrations = append(lintegrations.Integrations, l.Integrations...)
		if l.NextPageToken == "" {
			break
		}
	}

	errChan := make(chan error)
	workChan := make(chan integration, len(lintegrations.Integrations))

	fanOutWg := sync.WaitGroup{}
	fanInWg := sync.WaitGroup{}

	errs := []string{}
	fanInWg.Add(1)

	go func() {
		defer fanInWg.Done()
		for {
			newErr, ok := <-errChan
			if !ok {
				return
			}
			errs = append(errs, newErr.Error())
		}
	}()

	for i := 0; i < numConnections; i++ {
		fanOutWg.Add(1)
		go exportWorker(&fanOutWg, workChan, errChan)
	}

	for _, i := range lintegrations.Integrations {
		workChan <- i
	}

	close(workChan)
	fanOutWg.Wait()
	close(errChan)
	fanInWg.Wait()

	if len(errs) > 0 {
		return errors.New(strings.Join(errs, "\n"))
	}

	return nil
}