<commit_before>package blocks

import (
	"bytes"
	"encoding/json"
	"log"
	"net/http"
	"strings"
	"time"
)

// HTTPStream creates a persistent HTTP connection, emitting all messages from
// the stream into streamtools
func HTTPStream(b *Block) {

	type longHTTPRule struct {
		Endpoint string
		Auth     string
	}
	var rule *longHTTPRule

	var body bytes.Buffer
	// these are the possible delimiters
	d1 := []byte{125, 10} // this is }\n
	d2 := []byte{13, 10}  // this is CRLF

	client := &http.Client{}
	var res *http.Response

	for {
		select {
		case m := <-b.Routes["set_rule"]:
			if rule == nil {
				rule = &longHTTPRule{}
			}
			unmarshal(m, rule)
			req, err := http.NewRequest("GET", rule.Endpoint, nil)
			if err != nil {
				log.Fatal(err)
			}
			if len(rule.Auth) > 0 {
				req.SetBasicAuth(strings.Split(rule.Auth, ":")[0], strings.Split(rule.Auth, ":")[1])
			}
			res, err = client.Do(req)
			if err != nil {
				log.Fatal(err)
			}
			defer res.Body.Close()

		case r := <-b.Routes["get_rule"]:
			if rule == nil {
				marshal(r, &longHTTPRule{})
			} else {
				marshal(r, rule)
			}
		case msg := <-b.AddChan:
			updateOutChans(msg, b)
		case <-b.QuitChan:
			quit(b)
			return
		default:
			if res == nil {
				time.Sleep(500 * time.Millisecond)
				break
			}
			buffer := make([]byte, 5*1024)
			p, err := res.Body.Read(buffer)
			if err != nil {
				log.Fatal(err.Error())
			}
			// catch odd little buffers
			if p < 2 {
				break
			}
			body.Write(buffer[:p])
			if bytes.Equal(d1, buffer[p-2:p]) { // ended with }\n
				for _, blob := range bytes.Split(body.Bytes(), []byte{10}) { // split on new line in case there are multuple messages per buffer
					if len(blob) > 0 {
						var outMsg interface{}
						json.Unmarshal(blob, &outMsg)
						out := BMsg{
							Msg: outMsg,
						}<commit_msg>addressed Nik's comment about hte error<commit_after>package blocks

import (
	"bytes"
	"encoding/json"
	"log"
	"net/http"
	"strings"
	"time"
)

// HTTPStream creates a persistent HTTP connection, emitting all messages from
// the stream into streamtools
func HTTPStream(b *Block) {

	type longHTTPRule struct {
		Endpoint string
		Auth     string
	}
	var rule *longHTTPRule

	var body bytes.Buffer
	// these are the possible delimiters
	d1 := []byte{125, 10} // this is }\n
	d2 := []byte{13, 10}  // this is CRLF

	client := &http.Client{}
	var res *http.Response

	for {
		select {
		case m := <-b.Routes["set_rule"]:
			if rule == nil {
				rule = &longHTTPRule{}
			}
			unmarshal(m, rule)
			req, err := http.NewRequest("GET", rule.Endpoint, nil)
			if err != nil {
				log.Fatal(err)
			}
			if len(rule.Auth) > 0 {
				req.SetBasicAuth(strings.Split(rule.Auth, ":")[0], strings.Split(rule.Auth, ":")[1])
			}
			res, err = client.Do(req)
			if err != nil {
				log.Fatal(err)
			}
			defer res.Body.Close()

		case r := <-b.Routes["get_rule"]:
			if rule == nil {
				marshal(r, &longHTTPRule{})
			} else {
				marshal(r, rule)
			}
		case msg := <-b.AddChan:
			updateOutChans(msg, b)
		case <-b.QuitChan:
			quit(b)
			return
		default:
			if res == nil {
				time.Sleep(500 * time.Millisecond)
				break
			}
			buffer := make([]byte, 5*1024)
			p, err := res.Body.Read(buffer)
			if err != nil {
				log.Fatal(err.Error())
			}
			// catch odd little buffers
			if p < 2 {
				break
			}
			body.Write(buffer[:p])
			if bytes.Equal(d1, buffer[p-2:p]) { // ended with }\n
				for _, blob := range bytes.Split(body.Bytes(), []byte{10}) { // split on new line in case there are multuple messages per buffer
					if len(blob) > 0 {
						var outMsg interface{}
						err := json.Unmarshal(blob, &outMsg)
						if err != nil {
							log.Println("cannot unmarshal json")
							break
						}
						out := BMsg{
							Msg: outMsg,
						}
{'fraction_non_alphanumeric': 0.15278514588859415, 'fraction_numerical': 0.014854111405835544, 'mean_word_length': 2.5676442762535476, 'pattern_counts': {'":': 4, '<': 13, '<?xml version=': 0, '>': 7, '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': '7419968', 'n_tokens_mistral': 1729, 'n_tokens_neox': 1366, 'n_words': 371}