<commit_before>package garbage4

import (
	"sync"
	"time"
)

type Handler func(message []byte) error

type Channel struct {
	sync.RWMutex
	name    string
	topic   *Topic
	state   *State
	cond    *sync.Cond
	handler Handler
	waiting int
}

func newChannel(topic *Topic, name string) *Channel {
	c := &Channel{
		name:  name,
		topic: topic,
	}
	c.cond = &sync.Cond{L: &c.RWMutex}
	return c
}

func (c *Channel) Consume(handler Handler) {
	c.handler = handler
	// drain existing messages
	for {
		if message := c.topic.read(c); message != nil {
			c.handle(message)
		} else {
			break
		}
	}

	c.Lock()
	for {
		for c.waiting == 0 {
			c.cond.Wait()
		}
		c.Unlock()
		processed := 0
		for {
			message := c.topic.read(c)
			if message == nil {
				c.Lock()
				c.waiting -= processed
				break
			}
			if c.handle(message) {
				processed++
			}
		}
	}
}

func (c *Channel) notify() {
	c.Lock()
	c.waiting += 1
	c.Unlock()
	c.cond.Signal()
}

func (c *Channel) handle(message []byte) bool {
	if err := c.handler(message); err != nil {
		time.Sleep(time.Second) //todo: better
		return false
	}
	c.Lock()
	c.state.offset += uint32(len(message) + 4)
	c.Unlock()
	return true
}

func (c *Channel) isSegmentUsable(id uint64) bool {
	c.RLock()
	defer c.RUnlock()
	return c.state.segmentId <= id
}<commit_msg>push the 'backoff' up to calling code<commit_after>package garbage4

import (
	"sync"
	"time"
)

type Handler func(message []byte) error

type Channel struct {
	sync.RWMutex
	name    string
	topic   *Topic
	state   *State
	cond    *sync.Cond
	handler Handler
	waiting int
}

func newChannel(topic *Topic, name string) *Channel {
	c := &Channel{
		name:  name,
		topic: topic,
	}
	c.cond = &sync.Cond{L: &c.RWMutex}
	return c
}

func (c *Channel) Consume(handler Handler) {
	c.handler = handler
	// drain existing messages
	for {
		if message := c.topic.read(c); message != nil {
			c.handle(message)
		} else {
			break
		}
	}

	c.Lock()
	for {
		for c.waiting == 0 {
			c.cond.Wait()
		}
		c.Unlock()
		processed := 0
		for {
			message := c.topic.read(c)
			if message == nil {
				c.Lock()
				c.waiting -= processed
				break
			}
			if c.handle(message) {
				processed++
			}
		}
	}
}

func (c *Channel) notify() {
	c.Lock()
	c.waiting += 1
	c.Unlock()
	c.cond.Signal()
}

func (c *Channel) handle(message []byte) bool {
	if err := c.handler(message); err != nil {
		return false
	}
	c.Lock()
	c.state.offset += uint32(len(message) + 4)
	c.Unlock()
	return true
}

func (c *Channel) isSegmentUsable(id uint64) bool {
	c.RLock()
	defer c.RUnlock()
	return c.state.segmentId <= id
}
{'fraction_non_alphanumeric': 0.14947856315179606, 'fraction_numerical': 0.006952491309385863, 'mean_word_length': 2.8656716417910446, 'pattern_counts': {'":': 0, '<': 5, '<?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': '5055926', 'n_tokens_mistral': 1117, 'n_tokens_neox': 1017, 'n_words': 259}