convert MDNSClient actor to use closures for methods

We use a 'nil' closure to indicate shutdown.

nameserver/mdns_client.go

@@ -50,11 +50,11 @@ type inflightQuery struct {
 type MDNSClient struct {
 	// note unorthodox use of 'Server' class in client logic - using it to
 	// listen on a multicast socket and call callback whenever a message comes in.
-	listener  *dns.Server
-	conn      *net.UDPConn
-	addr      *net.UDPAddr
-	inflight  map[string]*inflightQuery
-	queryChan chan<- *MDNSInteraction
+	listener   *dns.Server
+	conn       *net.UDPConn
+	addr       *net.UDPAddr
+	inflight   map[string]*inflightQuery
+	actionChan chan<- MDNSAction
 }
 
 type mDNSQueryInfo struct {
@@ -90,9 +90,9 @@ func (c *MDNSClient) Start(ifi *net.Interface) error {
 	c.listener = &dns.Server{Unsafe: true, PacketConn: multicast, Handler: dns.HandlerFunc(handleMDNS)}
 	go c.listener.ActivateAndServe()
 
-	queryChan := make(chan *MDNSInteraction, MailboxSize)
-	c.queryChan = queryChan
-	go c.queryLoop(queryChan)
+	actionChan := make(chan MDNSAction, MailboxSize)
+	c.actionChan = actionChan
+	go c.actorLoop(actionChan)
 
 	return nil
 }
@@ -103,34 +103,79 @@ func LinkLocalMulticastListener(ifi *net.Interface) (net.PacketConn, error) {
 
 // ACTOR client API
 
-const (
-	CSendQuery = iota
-	CShutdown
-	CMessageReceived
-)
-
-type MDNSInteraction struct {
-	code       int
-	resultChan chan<- interface{}
-	payload    interface{}
-}
+type MDNSAction func()
 
 // Async
 func (c *MDNSClient) Shutdown() {
-	c.queryChan <- &MDNSInteraction{code: CShutdown}
+	c.actionChan <- nil
 }
 
 // Async
 func (c *MDNSClient) SendQuery(name string, querytype uint16, responseCh chan<- *Response) {
-	c.queryChan <- &MDNSInteraction{
-		code:    CSendQuery,
-		payload: mDNSQueryInfo{name, querytype, responseCh},
+	c.actionChan <- func() {
+		query, found := c.inflight[name]
+		if !found {
+			m := new(dns.Msg)
+			m.SetQuestion(name, querytype)
+			m.RecursionDesired = false
+			buf, err := m.Pack()
+			if err != nil {
+				responseCh <- &Response{Err: err}
+				close(responseCh)
+				return
+			}
+			query = &inflightQuery{
+				name: name,
+				id:   m.Id,
+			}
+			if _, err = c.conn.WriteTo(buf, c.addr); err != nil {
+				responseCh <- &Response{Err: err}
+				close(responseCh)
+				return
+			}
+			c.inflight[name] = query
+		}
+		info := &responseInfo{
+			ch:      responseCh,
+			timeout: time.Now().Add(mDNSTimeout),
+		}
+		// Invariant on responseInfos: they are in ascending order of
+		// timeout.  Since we use a fixed interval from Now(), this
+		// must be after all existing timeouts.
+		query.responseInfos = append(query.responseInfos, info)
 	}
 }
 
 // Async - called from dns library multiplexer
 func (c *MDNSClient) ResponseCallback(r *dns.Msg) {
-	c.queryChan <- &MDNSInteraction{code: CMessageReceived, payload: r}
+	c.actionChan <- func() {
+		for _, answer := range r.Answer {
+			var name string
+			var res *Response
+
+			switch rr := answer.(type) {
+			case *dns.A:
+				name = rr.Hdr.Name
+				res = &Response{Addr: rr.A}
+			case *dns.PTR:
+				name = rr.Hdr.Name
+				res = &Response{Name: rr.Ptr}
+			default:
+				return
+			}
+
+			if query, found := c.inflight[name]; found {
+				for _, resp := range query.responseInfos {
+					resp.ch <- res
+					close(resp.ch)
+				}
+				delete(c.inflight, name)
+			} else {
+				// We've received a response that didn't match a query
+				// Do we want to cache it?
+			}
+		}
+	}
 }
 
 // ACTOR server
@@ -164,25 +209,18 @@ func (c *MDNSClient) checkInFlightQueries() time.Duration {
 	return after
 }
 
-func (c *MDNSClient) queryLoop(queryChan <-chan *MDNSInteraction) {
+func (c *MDNSClient) actorLoop(actionChan <-chan MDNSAction) {
 	timer := time.NewTimer(MaxDuration)
 	run := func() { timer.Reset(c.checkInFlightQueries()) }
 	terminate := false
 	for !terminate {
 		select {
-		case query, ok := <-queryChan:
-			if !ok {
-				break
-			}
-			switch query.code {
-			case CShutdown:
+		case action := <-actionChan:
+			if action == nil {
 				c.listener.Shutdown()
 				terminate = true
-			case CSendQuery:
-				c.handleSendQuery(query.payload.(mDNSQueryInfo))
-				run()
-			case CMessageReceived:
-				c.handleResponse(query.payload.(*dns.Msg))
+			} else {
+				action()
 				run()
 			}
 		case <-timer.C:
@@ -197,64 +235,3 @@ func (c *MDNSClient) queryLoop(queryChan <-chan *MDNSInteraction) {
 		}
 	}
 }
-
-func (c *MDNSClient) handleSendQuery(q mDNSQueryInfo) {
-	query, found := c.inflight[q.name]
-	if !found {
-		m := new(dns.Msg)
-		m.SetQuestion(q.name, q.querytype)
-		m.RecursionDesired = false
-		buf, err := m.Pack()
-		if err != nil {
-			q.responseCh <- &Response{Err: err}
-			close(q.responseCh)
-			return
-		}
-		query = &inflightQuery{
-			name: q.name,
-			id:   m.Id,
-		}
-		if _, err = c.conn.WriteTo(buf, c.addr); err != nil {
-			q.responseCh <- &Response{Err: err}
-			close(q.responseCh)
-			return
-		}
-		c.inflight[q.name] = query
-	}
-	info := &responseInfo{
-		ch:      q.responseCh,
-		timeout: time.Now().Add(mDNSTimeout),
-	}
-	// Invariant on responseInfos: they are in ascending order of timeout.
-	// Since we use a fixed interval from Now(), this must be after all existing timeouts.
-	query.responseInfos = append(query.responseInfos, info)
-}
-
-func (c *MDNSClient) handleResponse(r *dns.Msg) {
-	for _, answer := range r.Answer {
-		var name string
-		var res *Response
-
-		switch rr := answer.(type) {
-		case *dns.A:
-			name = rr.Hdr.Name
-			res = &Response{Addr: rr.A}
-		case *dns.PTR:
-			name = rr.Hdr.Name
-			res = &Response{Name: rr.Ptr}
-		default:
-			return
-		}
-
-		if query, found := c.inflight[name]; found {
-			for _, resp := range query.responseInfos {
-				resp.ch <- res
-				close(resp.ch)
-			}
-			delete(c.inflight, name)
-		} else {
-			// We've received a response that didn't match a query
-			// Do we want to cache it?
-		}
-	}
-}