<commit_before>package ironmq

import (
	"bytes"
	"encoding/json"
	"errors"
	"fmt"
	"io/ioutil"
	"math/rand"
	"net/http"
	"net/http/httputil"
	"path"
	"sync"
	"time"
)

// Copied straight from Go's package rand since it's not exported.
type lockedSource struct {
	lk  sync.Mutex
	src rand.Source
}

func (r *lockedSource) Int63() (n int64) {
	r.lk.Lock()
	n = r.src.Int63()
	r.lk.Unlock()
	return
}

func (r *lockedSource) Seed(seed int64) {
	r.lk.Lock()
	r.src.Seed(seed)
	r.lk.Unlock()
}

var localRand = rand.New(&lockedSource{src: rand.NewSource(time.Now().Unix())})

// A Client contains an Iron.io project ID and a token for authentication.
type Client struct {
	Debug     bool
	cloud     *Cloud
	projectId string
	token     string
}

// NewClient returns a new Client using the given project ID and token.
// The network is not used during this call.
func NewClient(projectId, token string, cloud *Cloud) *Client {
	return &Client{projectId: projectId, token: token, cloud: cloud}
}

type Error struct {
	Status int
	Msg    string
}

var EmptyQueue = errors.New("queue is empty")

func (e *Error) Error() string { return fmt.Sprintf("Status %d: %s", e.Status, e.Msg) }

func (c *Client) req(method, endpoint string, body []byte, data interface{}) error {
	const apiVersion = "1"
	url := path.Join(c.cloud.host, apiVersion, "projects", c.projectId, endpoint)
	url = c.cloud.scheme + "://" + url + "?oauth=" + c.token
	req, err := http.NewRequest(method, url, bytes.NewBuffer(body))
	if err != nil {
		return err
	}
	if body != nil {
		req.Header.Set("Content-Type", "application/json")
		req.ContentLength = int64(len(body))
	}

	const maxRetries = 5
	tries := uint(0)
	var resp *http.Response
	for tries < maxRetries {
		resp, err = http.DefaultClient.Do(req)
		if err != nil {
			return err
		}
		// ELB sometimes returns this when load is increasing; we retry
		// with exponential backoff
		if resp.StatusCode == http.StatusServiceUnavailable {
			tries++
			// random delay between 0 and (4^tries*100) milliseconds
			pow := int64(1) << (2 * tries) * 100
			delayMs := time.Duration(localRand.Int63n(pow))
			time.Sleep(delayMs * time.Millisecond)
			req.Body = ioutil.NopCloser(bytes.NewBuffer(body))
			continue
		}
		break
	}

	if c.Debug {
		dump, err := httputil.DumpResponse(resp, true)
		if err != nil {
			fmt.Println("error dumping response:", err)
		} else {
			fmt.Printf("IRONMQ_GO_CLIENT:  %s\n", dump)<commit_msg>Work around EOF errors<commit_after>package ironmq

import (
	"bytes"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"io/ioutil"
	"math/rand"
	"net/http"
	"net/http/httputil"
	"path"
	"sync"
	"time"
)

// Copied straight from Go's package rand since it's not exported.
type lockedSource struct {
	lk  sync.Mutex
	src rand.Source
}

func (r *lockedSource) Int63() (n int64) {
	r.lk.Lock()
	n = r.src.Int63()
	r.lk.Unlock()
	return
}

func (r *lockedSource) Seed(seed int64) {
	r.lk.Lock()
	r.src.Seed(seed)
	r.lk.Unlock()
}

var localRand = rand.New(&lockedSource{src: rand.NewSource(time.Now().Unix())})

// A Client contains an Iron.io project ID and a token for authentication.
type Client struct {
	Debug     bool
	cloud     *Cloud
	projectId string
	token     string
}

// NewClient returns a new Client using the given project ID and token.
// The network is not used during this call.
func NewClient(projectId, token string, cloud *Cloud) *Client {
	return &Client{projectId: projectId, token: token, cloud: cloud}
}

type Error struct {
	Status int
	Msg    string
}

var EmptyQueue = errors.New("queue is empty")

func (e *Error) Error() string { return fmt.Sprintf("Status %d: %s", e.Status, e.Msg) }

func (c *Client) req(method, endpoint string, body []byte, data interface{}) error {
	const apiVersion = "1"
	url := path.Join(c.cloud.host, apiVersion, "projects", c.projectId, endpoint)
	url = c.cloud.scheme + "://" + url + "?oauth=" + c.token
	req, err := http.NewRequest(method, url, bytes.NewBuffer(body))
	if err != nil {
		return err
	}
	if body != nil {
		req.Header.Set("Content-Type", "application/json")
		req.ContentLength = int64(len(body))
	}

	const maxRetries = 5
	tries := uint(0)
	expectingEOF := false
	var resp *http.Response
	for tries < maxRetries {
		resp, err = http.DefaultClient.Do(req)
		if expectingEOF && err == io.EOF {
			continue
		}
		if err != nil {
			return err
		}
		// ELB sometimes returns this when load is increasing; we retry
		// with exponential backoff
		if resp.StatusCode == http.StatusServiceUnavailable {
			tries++
			// random delay between 0 and (4^tries*100) milliseconds
			pow := int64(1) << (2 * tries) * 100
			delayMs := time.Duration(localRand.Int63n(pow))
			time.Sleep(delayMs * time.Millisecond)
			req.Body = ioutil.NopCloser(bytes.NewBuffer(body))
			// the next retry is expected to fail; issue between Go's
			// core library and ELB
			expectingEOF = true
			continue
		}
		break
	}

	if c.Debug {
		dump, err := httputil.DumpResponse(resp, true)
		if err != nil {
			fmt.Println("error dumping response:", err)
		} else {
			fmt.Printf("IRONMQ_GO_CLIENT:  %s\n", dump)
{'fraction_non_alphanumeric': 0.13706183536825522, 'fraction_numerical': 0.01063410791650256, 'mean_word_length': 3.5798016230838594, 'pattern_counts': {'":': 2, '<': 9, '<?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': '1288509', 'n_tokens_mistral': 1940, 'n_tokens_neox': 1801, 'n_words': 524}