<commit_before>package blocks

import (
	"encoding/json"
	"log"
	"regexp"
)

type opFunc func(interface{}, interface{}) bool

var operators map[string]opFunc

// Filter queries a message for all values that match the given Path parameter
// and compares those values to a rule given an operator and a comparator. If
// any value passes the filter operation then the message is broadcast. If no
// value satisfies the filter operation the message is ignored.
//
// Filter is capable of traversing arrays that contain elements with and
// without keys. For example, if Path is set to
// 		foo.bar[]
// All elements within the "bar" array will be compared to the filter operation.
// In the case of
//		foo.bar[].baz
// Only the value of the "baz" keys within elements of the "bar" array will be
// used for the filter operation.
//
// There are four filter comparators: greater than "gt", less than "lt", equal
// to "eq" and subset of "subset".
//
// gt, lt, eq operations are available for values of a number type.
// eq, subset operations are avilable for values of a string type.
// eq operations are availble for value of a bool or null type.
func Filter(b *Block) {

	type filterRule struct {
		Operator   string
		Path       string
		Comparator interface{}
		Invert     bool
	}

	operators = make(map[string]opFunc)

	operators["eq"] = equals
	operators["gt"] = greaterthan
	operators["lt"] = lessthan
	operators["subset"] = subsetof
	operators["regex"] = regexmatch
	operators["keyin"] = keyin

	var rule *filterRule

	for {
		select {
		case msg := <-b.InChan:
			if rule == nil {
				break
			}

			values := getKeyValues(msg, rule.Path)
			for _, value := range values {
				if operators[rule.Operator](value, rule.Comparator) == !rule.Invert {
					broadcast(b.OutChans, msg)
					break
				}
			}

			if len(values) == 0 && rule.Invert {
				broadcast(b.OutChans, msg)
			}

		case msg := <-b.Routes["set_rule"]:
			if rule == nil {
				rule = &filterRule{}
			}
			// this is used to send back on the response chan
			var ruleRegexString string
			// we can't use the standard unmarshal(msg, rule) as we need to make
			// sure the regex compiles, if supplied.
			newRule := &filterRule{}
			err := json.Unmarshal(msg.Msg, &newRule)
			if err != nil {
				log.Println("found errors during unmarshalling")
				log.Println(err.Error())
				break
			}
			if newRule.Operator == "regex" {
				// regex is a bit of a special case
				c, ok := newRule.Comparator.(string)
				if !ok {
					log.Println("regex must be a string, not setting rule")
					break
				}
				r, err := regexp.Compile(c)
				if err != nil {
					log.Println("regex did not compile, not setting rule")
					log.Println(err.Error())
					break
				}
				rule = newRule
				rule.Comparator = r
				ruleRegexString = r.String()
			} else {
				// the simpler rules don't need any futzing
				rule = newRule
			}
			// send the rule back for the response
			out_rule := *rule
			if rule.Operator == "regex" {
				// we replace the regex in the outgoing rule with its string
				// representation so we can marshal it correctly.
				out_rule.Comparator = ruleRegexString
			}
			m, err := json.Marshal(out_rule)
			if err != nil {
				log.Println("could not marshal new rule")
				break
			}
			msg.ResponseChan <- m
		case msg := <-b.Routes["get_rule"]:<commit_msg>attempts to make filter less fragile<commit_after>package blocks

import (
	"encoding/json"
	"log"
	"regexp"
)

type opFunc func(interface{}, interface{}) bool

var operators map[string]opFunc

// Filter queries a message for all values that match the given Path parameter
// and compares those values to a rule given an operator and a comparator. If
// any value passes the filter operation then the message is broadcast. If no
// value satisfies the filter operation the message is ignored.
//
// Filter is capable of traversing arrays that contain elements with and
// without keys. For example, if Path is set to
// 		foo.bar[]
// All elements within the "bar" array will be compared to the filter operation.
// In the case of
//		foo.bar[].baz
// Only the value of the "baz" keys within elements of the "bar" array will be
// used for the filter operation.
//
// There are four filter comparators: greater than "gt", less than "lt", equal
// to "eq" and subset of "subset".
//
// gt, lt, eq operations are available for values of a number type.
// eq, subset operations are avilable for values of a string type.
// eq operations are availble for value of a bool or null type.
func Filter(b *Block) {

	type filterRule struct {
		Operator   string
		Path       string
		Comparator interface{}
		Invert     bool
	}

	operators = make(map[string]opFunc)

	operators["eq"] = equals
	operators["gt"] = greaterthan
	operators["lt"] = lessthan
	operators["subset"] = subsetof
	operators["regex"] = regexmatch
	operators["keyin"] = keyin

	var rule *filterRule

	for {
		select {
		case msg := <-b.InChan:
			if rule == nil {
				break
			}

			values := getKeyValues(msg, rule.Path)
			for _, value := range values {
				op, ok := operators[rule.Operator]
				if !ok {
					log.Println("specified operator does not exist")
				}
				break
				if op(value, rule.Comparator) == !rule.Invert {
					broadcast(b.OutChans, msg)
					break
				}
			}

			if len(values) == 0 && rule.Invert {
				broadcast(b.OutChans, msg)
			}

		case msg := <-b.Routes["set_rule"]:
			if rule == nil {
				rule = &filterRule{}
			}
			// this is used to send back on the response chan
			var ruleRegexString string
			// we can't use the standard unmarshal(msg, rule) as we need to make
			// sure the regex compiles, if supplied.
			newRule := &filterRule{}
			err := json.Unmarshal(msg.Msg, &newRule)
			if err != nil {
				log.Println("found errors during unmarshalling")
				log.Println(err.Error())
				m, _ := json.Marshal(rule)
				msg.ResponseChan <- m
				break
			}
			if _, ok := operators[newRule.Operator]; !ok {
				log.Println("invalid operator")
				m, _ := json.Marshal(rule)
				msg.ResponseChan <- m
				break
			}
			if newRule.Operator == "regex" {
				// regex is a bit of a special case
				c, ok := newRule.Comparator.(string)
				if !ok {
					log.Println("regex must be a string, not setting rule")
					m, _ := json.Marshal(rule)
					msg.ResponseChan <- m
					break
				}
				r, err := regexp.Compile(c)
				if err != nil {
					log.Println("regex did not compile, not setting rule")
					log.Println(err.Error())
					m, _ := json.Marshal(rule)
					msg.ResponseChan <- m
					break
				}
				rule = newRule
				rule.Comparator = r
				ruleRegexString = r.String()
			} else {
				// the simpler rules don't need any futzing
				rule = newRule
			}
			// send the rule back for the response
			out_rule := *rule
			if rule.Operator == "regex" {
				// we replace the regex in the outgoing rule with its string
				// representation so we can marshal it correctly.
				out_rule.Comparator = ruleRegexString
			}
			m, err := json.Marshal(out_rule)
			if err != nil {
				log.Println("could not marshal new rule")
				m, _ := json.Marshal(rule)
				msg.ResponseChan <- m
				break
			}
			msg.ResponseChan <- m
		case msg := <-b.Routes["get_rule"]:
{'fraction_non_alphanumeric': 0.1102460725705547, 'fraction_numerical': 0.00027804810232170165, 'mean_word_length': 2.7843240399789586, 'pattern_counts': {'":': 0, '<': 16, '<?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': '7413249', 'n_tokens_mistral': 2653, 'n_tokens_neox': 2195, 'n_words': 828}