<commit_before>package queue

import (
	"bytes"
	"io/ioutil"
	"log"
	"net/http"
	"os"
	"sync"
	"time"

	"github.com/StackExchange/tcollector/opentsdb"
)

var l = log.New(os.Stdout, "", log.LstdFlags)

type Queue struct {
	sync.Mutex
	host  string
	queue opentsdb.MultiDataPoint
	c     chan *opentsdb.DataPoint
}

// Creates and starts a new Queue.
func New(host string, c chan *opentsdb.DataPoint) *Queue {
	q := Queue{
		host: host,
		c:    c,
	}

	go func() {
		for dp := range c {
			q.Lock()
			q.queue = append(q.queue, dp)
			q.Unlock()
		}
	}()
	go q.send()
	return &q
}

var MAX_PERSEC = 50

func (q *Queue) send() {
	for {
		if len(q.queue) > 0 {
			q.Lock()
			i := len(q.queue)
			if i > MAX_PERSEC {
				i = MAX_PERSEC
			}
			sending := q.queue[:i]
			q.queue = q.queue[i:]
			go q.sendBatch(sending)
			q.Unlock()
		} else {
			time.Sleep(time.Second)
		}
	}
}

func (q *Queue) sendBatch(batch opentsdb.MultiDataPoint) {
	l.Println("sending", len(batch))
	b, err := batch.Json()
	if err != nil {
		l.Println(err)
		// bad JSON encoding, just give up
		return
	}
	resp, err := http.Post(q.host, "application/json", bytes.NewReader(b))
	if err != nil {
		l.Println(err)
	} else if resp.StatusCode != http.StatusNoContent {
		l.Println("RESP ERR", resp.Status)
		defer resp.Body.Close()
		body, err := ioutil.ReadAll(resp.Body)
		if err != nil {
			l.Println("ERR", err)
		}
		if len(body) > 0 {
			l.Println("ERR BODY", string(body))
		}
		l.Println("REQ BODY", string(b))
	}
}<commit_msg>cmd/scollector: Revert "Remove error handling"<commit_after>package queue

import (
	"bytes"
	"io/ioutil"
	"log"
	"net/http"
	"os"
	"sync"
	"time"

	"github.com/StackExchange/tcollector/opentsdb"
)

var l = log.New(os.Stdout, "", log.LstdFlags)

type Queue struct {
	sync.Mutex
	host  string
	queue opentsdb.MultiDataPoint
	c     chan *opentsdb.DataPoint
}

// Creates and starts a new Queue.
func New(host string, c chan *opentsdb.DataPoint) *Queue {
	q := Queue{
		host: host,
		c:    c,
	}

	go func() {
		for dp := range c {
			q.Lock()
			q.queue = append(q.queue, dp)
			q.Unlock()
		}
	}()
	go q.send()
	return &q
}

var MAX_PERSEC = 50

func (q *Queue) send() {
	for {
		if len(q.queue) > 0 {
			q.Lock()
			i := len(q.queue)
			if i > MAX_PERSEC {
				i = MAX_PERSEC
			}
			sending := q.queue[:i]
			q.queue = q.queue[i:]
			go q.sendBatch(sending)
			q.Unlock()
		} else {
			time.Sleep(time.Second)
		}
	}
}

func (q *Queue) sendBatch(batch opentsdb.MultiDataPoint) {
	l.Println("sending", len(batch))
	b, err := batch.Json()
	if err != nil {
		l.Println(err)
		// bad JSON encoding, just give up
		return
	}
	resp, err := http.Post(q.host, "application/json", bytes.NewReader(b))
	if err != nil {
		l.Println(err)
		goto Err
	}
	if resp.StatusCode != http.StatusNoContent {
		l.Println("RESP ERR", resp.Status)
		defer resp.Body.Close()
		body, err := ioutil.ReadAll(resp.Body)
		if err != nil {
			l.Println("ERR", err)
		}
		if len(body) > 0 {
			l.Println("ERR BODY", string(body))
		}
		l.Println("REQ BODY", string(b))
		goto Err
	}
	return
Err:
	l.Println("error, restoring", len(batch))
	for _, dp := range batch {
		q.c <- dp
	}
}
{'fraction_non_alphanumeric': 0.1649746192893401, 'fraction_numerical': 0.0025380710659898475, 'mean_word_length': 2.946182728410513, 'pattern_counts': {'":': 0, '<': 4, '<?xml version=': 0, '>': 9, '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': '4095513', 'n_tokens_mistral': 1457, 'n_tokens_neox': 1330, 'n_words': 306}