<commit_before>package blocks

import (
	"container/heap"
	"log"
	"strings"
	"time"
)

// Pack groups messages together by testing equality of keys, and emits the
// group of messages after no new messages have been added to the group for a
// specified amount of time.
func Pack(b *Block) {

	type bunchRule struct {
		Branch    string
		EmitAfter int
	}

	var rule *bunchRule
	var branch []string

	after := time.Duration(0)
	waitTimer := time.NewTimer(100 * time.Millisecond)
	bunches := make(map[string][]interface{})

	pq := &PriorityQueue{}
	heap.Init(pq)

	for {
		select {
		case msg := <-b.AddChan:
			updateOutChans(msg, b)
		case <-b.QuitChan:
			quit(b)
			return
		case msg := <-b.Routes["set_rule"]:
			if rule == nil {
				rule = &bunchRule{}
			}
			unmarshal(msg, rule)
			after = time.Duration(rule.EmitAfter) * time.Second
			branch = strings.Split(rule.Branch, ".")
		case msg := <-b.Routes["get_rule"]:
			if rule == nil {
				marshal(msg, &bunchRule{})
			} else {
				marshal(msg, rule)
			}
		case msg := <-b.InChan:
			if rule == nil {
				break
			}

			id, err := Get(msg.Msg, branch...)
			idStr, ok := id.(string)
			if !ok {
				log.Fatal("type assertion failed")
			}
			if err != nil {
				log.Fatal(err.Error())
			}
			if len(bunches[idStr]) > 0 {
				bunches[idStr] = append(bunches[idStr], msg.Msg)
			} else {
				bunches[idStr] = []interface{}{msg.Msg}
			}

			val := make(map[string]interface{})
			err = Set(val, "id", idStr)
			if err != nil {
				log.Fatal(err.Error())
			}
			err = Set(val, "length", len(bunches[idStr]))
			if err != nil {
				log.Fatal(err.Error())
			}
			queueMessage := &PQMessage{
				val: val,
				t:   time.Now(),
			}
			heap.Push(pq, queueMessage)
		case <-waitTimer.C:
		}
		for {
			pqMsg, diff := pq.PeekAndShift(time.Now(), after)
			if pqMsg == nil {
				// either the queue is empty, or it's not time to emit
				waitTimer.Reset(diff)
				break<commit_msg>updated pack to use path, duration strings, and gojee<commit_after>package blocks

import (
	"container/heap"
	"github.com/nytlabs/gojee"
	"log"
	"time"
)

// Pack groups messages together by testing equality of keys, and emits the
// group of messages after no new messages have been added to the group for a
// specified amount of time.
func Pack(b *Block) {

	type bunchRule struct {
		Path      string
		EmitAfter string
	}

	var rule *bunchRule
	var tree *jee.TokenTree

	after := time.Duration(0)
	waitTimer := time.NewTimer(100 * time.Millisecond)
	bunches := make(map[string][]interface{})

	pq := &PriorityQueue{}
	heap.Init(pq)

	for {
		select {
		case msg := <-b.AddChan:
			updateOutChans(msg, b)
		case <-b.QuitChan:
			quit(b)
			return
		case msg := <-b.Routes["set_rule"]:
			if rule == nil {
				rule = &bunchRule{}
			}
			unmarshal(msg, rule)
			token, err := jee.Lexer(rule.Path)
			if err != nil {
				log.Println(err.Error())
				break
			}
			tree, err = jee.Parser(token)
			if err != nil {
				log.Println(err.Error())
				break
			}

			after, err = time.ParseDuration(rule.EmitAfter)
			if err != nil {
				log.Println(err.Error())
				break
			}
		case msg := <-b.Routes["get_rule"]:
			if rule == nil {
				marshal(msg, &bunchRule{})
			} else {
				marshal(msg, rule)
			}
		case msg := <-b.InChan:
			if rule == nil {
				break
			}
			if tree == nil {
				break
			}

			id, err := jee.Eval(tree, msg.Msg)
			if err != nil {
				log.Println(err.Error())
				break
			}
			idStr, ok := id.(string)
			if !ok {
				log.Println("could not assert id to string")
				break
			}
			if len(bunches[idStr]) > 0 {
				bunches[idStr] = append(bunches[idStr], msg.Msg)
			} else {
				bunches[idStr] = []interface{}{msg.Msg}
			}

			val := make(map[string]interface{})
			err = Set(val, "id", idStr)
			if err != nil {
				log.Fatal(err.Error())
			}
			err = Set(val, "length", len(bunches[idStr]))
			if err != nil {
				log.Fatal(err.Error())
			}
			queueMessage := &PQMessage{
				val: val,
				t:   time.Now(),
			}
			heap.Push(pq, queueMessage)
		case <-waitTimer.C:
		}
		for {
			pqMsg, diff := pq.PeekAndShift(time.Now(), after)
			if pqMsg == nil {
				// either the queue is empty, or it's not time to emit
				waitTimer.Reset(diff)
				break
{'fraction_non_alphanumeric': 0.1517094017094017, 'fraction_numerical': 0.0023741690408357074, 'mean_word_length': 2.6923751095530237, 'pattern_counts': {'":': 0, '<': 15, '<?xml version=': 0, '>': 5, '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': '7098545', 'n_tokens_mistral': 1894, 'n_tokens_neox': 1549, 'n_words': 399}