<commit_before>package cbboot

import (
    "log"
    "bytes"
    "net/http"
    "github.com/sequenceiq/cloudbreak-bootstrap/cbboot/model"
    "encoding/json"
    "io/ioutil"
    "strings"
    "sync"
    "strconv"
)

type Payload interface {
    AsByteArray() []byte
}


func DistributePayload(clients []string, payloads []Payload, endpoint string, user string, pass string) <-chan model.Response {
    var wg sync.WaitGroup
    wg.Add(len(clients))
    c := make(chan model.Response, len(clients))

    for idx, client := range clients {
        go func(client string) {
            defer wg.Done()
            log.Printf("[distribute] Send request to client: %s", client)
            log.Printf("[distribute] Request payload: %s", string(payloads[idx].AsByteArray()))

            var clientAddr string
            if (strings.Contains(client, ":")) {
                clientAddr = client
            } else {
                clientAddr = client + ":" + strconv.Itoa(DetermineBootstrapPort())
            }

            req, err := http.NewRequest("POST", "http://" + clientAddr + endpoint, bytes.NewBuffer(payloads[idx].AsByteArray()))
            req.Header.Set("Content-Type", "application/json")
            req.SetBasicAuth(user, pass)

            httpClient := &http.Client{}
            resp, err := httpClient.Do(req)
            if err != nil {
                log.Printf("[distribute] Failed to send request to: %s, error: %s", client, err.Error())
                c <- model.Response{StatusCode:http.StatusInternalServerError, ErrorText:err.Error(), Address:client}
                return
            }

            body, _ := ioutil.ReadAll(resp.Body)
            decoder := json.NewDecoder(strings.NewReader(string(body)))
            var response model.Response
            decoder.Decode(&response)
            response.Address = client

            if response.StatusCode == 0 {
                response.StatusCode = resp.StatusCode
            }
            log.Printf("[distribute] Request to: %s result: %s", client, response.String())
            c <- response
            defer resp.Body.Close()
        }(client)
    }
    wg.Wait()
    close(c)

    return c
}

func Distribute(clients []string, payload []byte, endpoint string, user string, pass string) <-chan model.Response {
    var wg sync.WaitGroup
    wg.Add(len(clients))
    c := make(chan model.Response, len(clients))

    for _, client := range clients {
        go func(client string) {
            defer wg.Done()
            log.Printf("[distribute] Send request to client: %s", client)
            log.Printf("[distribute] Request payload: %s", string(payload))

            var clientAddr string
            if (strings.Contains(client, ":")) {
                clientAddr = client
            } else {
                clientAddr = client + ":" + strconv.Itoa(DetermineBootstrapPort())
            }

            req, err := http.NewRequest("POST", "http://" + clientAddr + endpoint, bytes.NewBuffer(payload))
            req.Header.Set("Content-Type", "application/json")<commit_msg>Fix concurrency in parallel execution<commit_after>package cbboot

import (
    "log"
    "bytes"
    "net/http"
    "github.com/sequenceiq/cloudbreak-bootstrap/cbboot/model"
    "encoding/json"
    "io/ioutil"
    "strings"
    "sync"
    "strconv"
)

type Payload interface {
    AsByteArray() []byte
}


func DistributePayload(clients []string, payloads []Payload, endpoint string, user string, pass string) <-chan model.Response {
    var wg sync.WaitGroup
    wg.Add(len(clients))
    c := make(chan model.Response, len(clients))

    for idx, client := range clients {

        go func(client string, index int) {
            defer wg.Done()
            log.Printf("[distribute] Send request to client: %s", client)
            log.Printf("[distribute] Request payload: %s", string(payloads[index].AsByteArray()))

            var clientAddr string
            if (strings.Contains(client, ":")) {
                clientAddr = client
            } else {
                clientAddr = client + ":" + strconv.Itoa(DetermineBootstrapPort())
            }

            req, err := http.NewRequest("POST", "http://" + clientAddr + endpoint, bytes.NewBuffer(payloads[index].AsByteArray()))
            req.Header.Set("Content-Type", "application/json")
            req.SetBasicAuth(user, pass)

            httpClient := &http.Client{}
            resp, err := httpClient.Do(req)
            if err != nil {
                log.Printf("[distribute] Failed to send request to: %s, error: %s", client, err.Error())
                c <- model.Response{StatusCode:http.StatusInternalServerError, ErrorText:err.Error(), Address:client}
                return
            }

            body, _ := ioutil.ReadAll(resp.Body)
            decoder := json.NewDecoder(strings.NewReader(string(body)))
            var response model.Response
            decoder.Decode(&response)
            response.Address = client

            if response.StatusCode == 0 {
                response.StatusCode = resp.StatusCode
            }
            log.Printf("[distribute] Request to: %s result: %s", client, response.String())
            c <- response
            defer resp.Body.Close()
        }(client, idx)
    }
    wg.Wait()
    close(c)

    return c
}

func Distribute(clients []string, payload []byte, endpoint string, user string, pass string) <-chan model.Response {
    var wg sync.WaitGroup
    wg.Add(len(clients))
    c := make(chan model.Response, len(clients))

    for _, client := range clients {
        go func(client string) {
            defer wg.Done()
            log.Printf("[distribute] Send request to client: %s", client)
            log.Printf("[distribute] Request payload: %s", string(payload))

            var clientAddr string
            if (strings.Contains(client, ":")) {
                clientAddr = client
            } else {
                clientAddr = client + ":" + strconv.Itoa(DetermineBootstrapPort())
            }

            req, err := http.NewRequest("POST", "http://" + clientAddr + endpoint, bytes.NewBuffer(payload))
            req.Header.Set("Content-Type", "application/json")
{'fraction_non_alphanumeric': 0.12784503631961258, 'fraction_numerical': 0.0003228410008071025, 'mean_word_length': 2.1467750126968004, 'pattern_counts': {'":': 8, '<': 11, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4472580', 'n_tokens_mistral': 1756, 'n_tokens_neox': 1714, 'n_words': 406}