diff --git a/main.go b/main.go index ad2ab95b..757242b7 100644 --- a/main.go +++ b/main.go @@ -29,7 +29,7 @@ func handleMDNS(w dns.ResponseWriter, r *dns.Msg) { // Here we have to handle both requests incoming, responses to queries // we sent out, and responses to queries that other nodes sent if len(r.Answer) > 0 { - mdnsClient.HandleResponse(r) + mdnsClient.ResponseCallback(r) } else if len(r.Question) > 0 { q := r.Question[0] ip, err := zone.MatchLocal(q.Name) @@ -131,6 +131,7 @@ func main() { if err != nil { log.Fatal(err) } + mdnsClient.Start() go dns.ActivateAndServe(nil, conn, MDNSServeMux) diff --git a/server/mdns_client.go b/server/mdns_client.go index 7a813288..36f40211 100644 --- a/server/mdns_client.go +++ b/server/mdns_client.go @@ -4,7 +4,6 @@ import ( "github.com/miekg/dns" "log" "net" - "sync" ) // Portions of this code taken from github.com/armon/mdns @@ -41,10 +40,16 @@ type inflightQuery struct { } type MDNSClient struct { - sync.RWMutex - conn *net.UDPConn - addr *net.UDPAddr - inflight map[string]*inflightQuery + conn *net.UDPConn + addr *net.UDPAddr + inflight map[string]*inflightQuery + queryChan chan<- *MDNSInteraction +} + +type mDNSQueryInfo struct { + name string + querytype uint16 + responseCh chan<- *ResponseA } func NewMDNSClient() (*MDNSClient, error) { @@ -60,34 +65,17 @@ func NewMDNSClient() (*MDNSClient, error) { return retval, nil } +func (c *MDNSClient) Start() { + queryChan := make(chan *MDNSInteraction, 4) + c.queryChan = queryChan + go c.queryLoop(queryChan) +} + func LinkLocalMulticastListener(ifi *net.Interface) (net.PacketConn, error) { conn, err := net.ListenMulticastUDP("udp", ifi, ipv4Addr) return conn, err } -func (c *MDNSClient) SendQuery(name string, querytype uint16, responseCh chan<- *ResponseA) error { - m := new(dns.Msg) - m.SetQuestion(name, querytype) - m.RecursionDesired = false - - c.Lock() - query, found := c.inflight[name] - if !found { - query = &inflightQuery{name: name, Id: m.Id} - c.inflight[name] = query - } - info := &responseInfo{ch: responseCh} - query.responseInfos = append(query.responseInfos, info) - c.Unlock() - - buf, err := m.Pack() - if err != nil { - return err - } - _, err = c.conn.WriteTo(buf, c.addr) - return err -} - func (c *MDNSClient) SendResponse(m *dns.Msg) error { buf, err := m.Pack() if err != nil { @@ -97,12 +85,100 @@ func (c *MDNSClient) SendResponse(m *dns.Msg) error { return err } +// ACTOR client API + +const ( + CSendQuery = iota + CShutdown = iota + CIsInflightQuery = iota + CMessageReceived = iota +) + +type Interaction struct { + code int + resultChan chan<- interface{} +} + +type MDNSInteraction struct { + Interaction + payload interface{} +} + +// Async +func (c *MDNSClient) Shutdown() { + c.queryChan <- &MDNSInteraction{Interaction: Interaction{code: CShutdown}} +} + +// Async +func (c *MDNSClient) SendQuery(name string, querytype uint16, responseCh chan<- *ResponseA) { + c.queryChan <- &MDNSInteraction{Interaction: Interaction{code: CSendQuery}, payload: mDNSQueryInfo{name, querytype, responseCh}} +} + +// Sync func (c *MDNSClient) IsInflightQuery(m *dns.Msg) bool { + resultChan := make(chan interface{}) + c.queryChan <- &MDNSInteraction{Interaction: Interaction{code: CIsInflightQuery, resultChan: resultChan}, payload: m} + result := <-resultChan + return result.(bool) +} + +// Async - called from dns library multiplexer +func (c *MDNSClient) ResponseCallback(r *dns.Msg) { + c.queryChan <- &MDNSInteraction{Interaction: Interaction{code: CMessageReceived}, payload: r} +} + +// ACTOR server + +func (c *MDNSClient) queryLoop(queryChan <-chan *MDNSInteraction) { + var err error + terminate := false + for !terminate { + if err != nil { + log.Printf("encountered error", err) + break + } + query, ok := <-queryChan + if !ok { + break + } + switch query.code { + case CShutdown: + terminate = true + case CSendQuery: + err = c.handleSendQuery(query.payload.(mDNSQueryInfo)) + case CIsInflightQuery: + query.resultChan <- c.handleIsInflightQuery(query.payload.(*dns.Msg)) + case CMessageReceived: + c.handleResponse(query.payload.(*dns.Msg)) + } + } + // handle shutdown here +} + +func (c *MDNSClient) handleSendQuery(q mDNSQueryInfo) error { + m := new(dns.Msg) + m.SetQuestion(q.name, q.querytype) + m.RecursionDesired = false + query, found := c.inflight[q.name] + if !found { + query = &inflightQuery{name: q.name, Id: m.Id} + c.inflight[q.name] = query + } + info := &responseInfo{ch: q.responseCh} + query.responseInfos = append(query.responseInfos, info) + + buf, err := m.Pack() + if err != nil { + return err + } + _, err = c.conn.WriteTo(buf, c.addr) + return err +} + +func (c *MDNSClient) handleIsInflightQuery(m *dns.Msg) bool { if len(m.Question) == 1 { q := m.Question[0] if q.Qtype == dns.TypeA { - c.Lock() - defer c.Unlock() if query, found := c.inflight[q.Name]; found { if query.Id == m.Id { return true @@ -113,11 +189,10 @@ func (c *MDNSClient) IsInflightQuery(m *dns.Msg) bool { return false } -func (c *MDNSClient) HandleResponse(r *dns.Msg) { +func (c *MDNSClient) handleResponse(r *dns.Msg) { for _, answer := range r.Answer { switch rr := answer.(type) { case *dns.A: - c.Lock() if query, found := c.inflight[rr.Hdr.Name]; found { for _, resp := range query.responseInfos { resp.ch <- &ResponseA{Name: rr.Hdr.Name, Addr: rr.A} @@ -128,7 +203,6 @@ func (c *MDNSClient) HandleResponse(r *dns.Msg) { // We've received a response that didn't match a query // Do we want to cache it? } - c.Unlock() } } }