<commit_before>package mesh

import (
	"fmt"
)

type GossipData interface {
	Encode() [][]byte
	Merge(GossipData) GossipData
}

type Gossip interface {
	// specific message from one peer to another
	// intermediate peers relay it using unicast topology.
	GossipUnicast(dstPeerName PeerName, msg []byte) error
	// send gossip to every peer, relayed using broadcast topology.
	GossipBroadcast(update GossipData) error
}

type Gossiper interface {
	OnGossipUnicast(sender PeerName, msg []byte) error
	// merge received data into state and return a representation of
	// the received data, for further propagation
	OnGossipBroadcast(sender PeerName, update []byte) (GossipData, error)
	// return state of everything we know; gets called periodically
	Gossip() GossipData
	// merge received data into state and return "everything new I've
	// just learnt", or nil if nothing in the received data was new
	OnGossip(update []byte) (GossipData, error)
}

// Accumulates GossipData that needs to be sent to one destination,
// and sends it when possible.
type GossipSender struct {
	send func(GossipData)
	cell chan GossipData
	// for testing
	sent    bool
	flushch chan chan bool
}

func NewGossipSender(send func(GossipData)) *GossipSender {
	cell := make(chan GossipData, 1)
	flushch := make(chan chan bool)
	s := &GossipSender{send: send, cell: cell, flushch: flushch}
	go s.run()
	return s
}

func (s *GossipSender) run() {
	for {
		select {
		case pending := <-s.cell:
			if pending == nil { // receive zero value when chan is closed
				return
			}
			s.send(pending)
			s.sent = true
		case ch := <-s.flushch:
			// send anything pending, then reply back whether we sent
			// anything since previous flush
			select {
			case pending := <-s.cell:
				s.send(pending)
				s.sent = true
			default:
			}
			ch <- s.sent
			s.sent = false
		}
	}
}

func (s *GossipSender) Send(data GossipData) {
	// NB: this must not be invoked concurrently
	select {
	case pending := <-s.cell:
		s.cell <- pending.Merge(data)
	default:
		s.cell <- data
	}
}

func (s *GossipSender) Stop() {<commit_msg>refactor: no need for GossipSender.sent<commit_after>package mesh

import (
	"fmt"
)

type GossipData interface {
	Encode() [][]byte
	Merge(GossipData) GossipData
}

type Gossip interface {
	// specific message from one peer to another
	// intermediate peers relay it using unicast topology.
	GossipUnicast(dstPeerName PeerName, msg []byte) error
	// send gossip to every peer, relayed using broadcast topology.
	GossipBroadcast(update GossipData) error
}

type Gossiper interface {
	OnGossipUnicast(sender PeerName, msg []byte) error
	// merge received data into state and return a representation of
	// the received data, for further propagation
	OnGossipBroadcast(sender PeerName, update []byte) (GossipData, error)
	// return state of everything we know; gets called periodically
	Gossip() GossipData
	// merge received data into state and return "everything new I've
	// just learnt", or nil if nothing in the received data was new
	OnGossip(update []byte) (GossipData, error)
}

// Accumulates GossipData that needs to be sent to one destination,
// and sends it when possible.
type GossipSender struct {
	send    func(GossipData)
	cell    chan GossipData
	flushch chan chan bool // for testing
}

func NewGossipSender(send func(GossipData)) *GossipSender {
	cell := make(chan GossipData, 1)
	flushch := make(chan chan bool)
	s := &GossipSender{send: send, cell: cell, flushch: flushch}
	go s.run()
	return s
}

func (s *GossipSender) run() {
	sent := false
	for {
		select {
		case pending := <-s.cell:
			if pending == nil { // receive zero value when chan is closed
				return
			}
			s.send(pending)
			sent = true
		case ch := <-s.flushch: // for testing
			// send anything pending, then reply back whether we sent
			// anything since previous flush
			select {
			case pending := <-s.cell:
				s.send(pending)
				sent = true
			default:
			}
			ch <- sent
			sent = false
		}
	}
}

func (s *GossipSender) Send(data GossipData) {
	// NB: this must not be invoked concurrently
	select {
	case pending := <-s.cell:
		s.cell <- pending.Merge(data)
	default:
		s.cell <- data
	}
}

func (s *GossipSender) Stop() {
{'fraction_non_alphanumeric': 0.09935821250297124, 'fraction_numerical': 0.0004753981459472308, 'mean_word_length': 3.554112554112554, 'pattern_counts': {'":': 0, '<': 17, '<?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': '5747283', 'n_tokens_mistral': 1456, 'n_tokens_neox': 1323, 'n_words': 490}