<commit_before>package zeroconf

import (
	"fmt"
	"log"
	"net"

	dns "github.com/miekg/godns"
)

var (
	Registry = newRegistry()
)



func openIPv4Socket() *net.UDPConn {
	conn, err := net.ListenUDP("udp4", &net.UDPAddr{
		IP:   net.IPv4zero,
		Port: 5353,
	})
	if err != nil {
		log.Fatal(err)
	}
	return conn
}

func init() {
	go Registry.mainloop()

	socket := openIPv4Socket()
	if err := socket.JoinGroup(nil, net.IPv4(224, 0, 0, 251)); err != nil {
		log.Fatal(err)
	}

	listener := &listener{
		socket:   socket,
		registry: Registry,
	}
	go listener.mainloop()

	// simple logger
	go func() {
		for op := range Registry.Subscribe() {
			switch op.Op {
			case Add:
				log.Printf("Add: %#v", op.Service)
			case Remove:
				log.Printf("Remove: %#v", op.Service)
			}
		}
	}()
}

func mcastInterfaces() []net.Interface {
	ifaces := make([]net.Interface, 0)
	interfaces, err := net.Interfaces()
	if err != nil {
		log.Fatal(err)
	}
	for _, i := range interfaces {
		if isMulticast(i) {
			fmt.Printf("%#v\n", i)
			ifaces = append(ifaces, i)
		}
	}
	return ifaces
}

func isMulticast(i net.Interface) bool {
	return (i.Flags&net.FlagUp > 0) && (i.Flags&net.FlagMulticast > 0)
}

type Service struct {
	Type      string
	Name      string
	Domain    string
	Interface *net.Interface
	Address   net.Addr
}

type op int

const (
	Add    op = 1
	Remove op = 2
)

type Operation struct {
	Op      op
	Service *Service
}

type registry struct {
	ops         chan Operation
	subscribe   chan *subscription
	services    []*Service
	subscribers []*subscription
}

// TODO registry is not an exported type, should it be?
func newRegistry() *registry {
	return &registry{
                ops:         make(chan Operation),
		subscribe:	make(chan *subscription),
                services:    nil,
                subscribers: nil,
        }
}

func (r *registry) Add(service *Service) {
	r.ops <- Operation{
		Op:      Add,
		Service: service,
	}
}

func (r *registry) Remove(service *Service) {
	r.ops <- Operation{
		Op:      Remove,
		Service: service,
	}
}

// TODO subscribe should take a *Query
func (r *registry) Subscribe() chan Operation {
	s := &subscription {
		make(chan Operation),
	}
	r.subscribe <- s
	return s.c
}

func (r *registry) mainloop() {
	for {
		select {
		case op := <-r.ops:
			switch op.Op {
			case Add:
				r.addService(op.Service)
			}
			r.notifySubscribers(op)
		}
	}
}

func (r *registry) addService(service *Service) {
	r.services = append(r.services, service)
}

func (r *registry) notifySubscribers(op Operation) {
	for i, _ := range r.subscribers {
		r.subscribers[i].notify(op)
	}
}

type subscription struct {
	c chan Operation
}

func (s *subscription) notify(op Operation) {
	s.c <- op // TODO use non blocking send in case reciver is full
}

type listener struct {
	socket   *net.UDPConn
	registry *registry
}

func (l *listener) mainloop() {<commit_msg>Process Subscribe() requests<commit_after>package zeroconf

import (
	"fmt"
	"log"
	"net"

	dns "github.com/miekg/godns"
)

var (
	Registry = newRegistry()
)

func openIPv4Socket() *net.UDPConn {
	conn, err := net.ListenUDP("udp4", &net.UDPAddr{
		IP:   net.IPv4zero,
		Port: 5353,
	})
	if err != nil {
		log.Fatal(err)
	}
	return conn
}

func init() {
	go Registry.mainloop()

	socket := openIPv4Socket()
	if err := socket.JoinGroup(nil, net.IPv4(224, 0, 0, 251)); err != nil {
		log.Fatal(err)
	}

	listener := &listener{
		socket:   socket,
		registry: Registry,
	}
	go listener.mainloop()

	// simple logger
	go func() {
		for op := range Registry.Subscribe() {
			switch op.Op {
			case Add:
				log.Printf("Add: %#v", op.Service)
			case Remove:
				log.Printf("Remove: %#v", op.Service)
			}
		}
	}()
}

func mcastInterfaces() []net.Interface {
	ifaces := make([]net.Interface, 0)
	interfaces, err := net.Interfaces()
	if err != nil {
		log.Fatal(err)
	}
	for _, i := range interfaces {
		if isMulticast(i) {
			fmt.Printf("%#v\n", i)
			ifaces = append(ifaces, i)
		}
	}
	return ifaces
}

func isMulticast(i net.Interface) bool {
	return (i.Flags&net.FlagUp > 0) && (i.Flags&net.FlagMulticast > 0)
}

type Service struct {
	Type      string
	Name      string
	Domain    string
	Interface *net.Interface
	Address   net.Addr
}

type op int

const (
	Add    op = 1
	Remove op = 2
)

type Operation struct {
	Op      op
	Service *Service
}

type registry struct {
	ops         chan Operation
	subscribe   chan *subscription
	services    []*Service
	subscribers []*subscription
}

// TODO registry is not an exported type, should it be?
func newRegistry() *registry {
	return &registry{
                ops:         make(chan Operation),
		subscribe:	make(chan *subscription),
                services:    nil,
                subscribers: nil,
        }
}

func (r *registry) Add(service *Service) {
	r.ops <- Operation{
		Op:      Add,
		Service: service,
	}
}

func (r *registry) Remove(service *Service) {
	r.ops <- Operation{
		Op:      Remove,
		Service: service,
	}
}

// TODO subscribe should take a *Query
func (r *registry) Subscribe() chan Operation {
	s := &subscription {
		make(chan Operation),
	}
	r.subscribe <- s
	return s.c
}

func (r *registry) mainloop() {
	for {
		select {
		case op := <-r.ops:
			switch op.Op {
			case Add:
				r.addService(op.Service)
			}
			r.notifySubscribers(op)
		case sub := <- r.subscribe:
			r.subscribers = append(r.subscribers, sub)
		}
	}
}

func (r *registry) addService(service *Service) {
	r.services = append(r.services, service)
}

func (r *registry) notifySubscribers(op Operation) {
	for i, _ := range r.subscribers {
		r.subscribers[i].notify(op)
	}
}

type subscription struct {
	c chan Operation
}

func (s *subscription) notify(op Operation) {
	s.c <- op // TODO use non blocking send in case reciver is full
}

type listener struct {
	socket   *net.UDPConn
	registry *registry
}

func (l *listener) mainloop() {
{'fraction_non_alphanumeric': 0.13596864883285056, 'fraction_numerical': 0.0074970182313852446, 'mean_word_length': 2.9264214046822743, 'pattern_counts': {'":': 0, '<': 14, '<?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': '46160', 'n_tokens_mistral': 2321, 'n_tokens_neox': 2146, 'n_words': 576}