Merge pull request #482 from rade/475-simplify-actors

simplify actors

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?
-		}
-	}
-}

router/connection.go

@@ -59,14 +59,11 @@ type LocalConnection struct {
 	Decryptor         Decryptor
 	Router            *Router
 	uid               uint64
-	queryChan         chan<- *ConnectionInteraction
-	finished          <-chan struct{} // closed to signal that queryLoop has finished
+	actionChan        chan<- ConnectionAction
+	finished          <-chan struct{} // closed to signal that actorLoop has finished
 }
 
-type ConnectionInteraction struct {
-	Interaction
-	payload interface{}
-}
+type ConnectionAction func() error
 
 func NewRemoteConnection(from, to *Peer, tcpAddr string, outbound bool, established bool) *RemoteConnection {
 	return &RemoteConnection{
@@ -118,11 +115,11 @@ func NewLocalConnection(connRemote *RemoteConnection, tcpConn *net.TCPConn, udpA
 // Async. Does not return anything. If the connection is successful,
 // it will end up in the local peer's connections map.
 func (conn *LocalConnection) Start(acceptNewPeer bool) {
-	queryChan := make(chan *ConnectionInteraction, ChannelSize)
-	conn.queryChan = queryChan
+	actionChan := make(chan ConnectionAction, ChannelSize)
+	conn.actionChan = actionChan
 	finished := make(chan struct{})
 	conn.finished = finished
-	go conn.run(queryChan, finished, acceptNewPeer)
+	go conn.run(actionChan, finished, acceptNewPeer)
 }
 
 func (conn *LocalConnection) BreakTie(dupConn Connection) ConnectionTieBreak {
@@ -182,29 +179,21 @@ func (conn *LocalConnection) SendProtocolMsg(m ProtocolMsg) {
 	}
 }
 
-// ACTOR client API
+// ACTOR methods
 
-const (
-	CSetEstablished = iota
-	CReceivedHeartbeat
-	CShutdown
-)
-
-// Send an actor request to the queryLoop, but don't block if queryLoop has exited
-// - see http://blog.golang.org/pipelines for pattern
-func (conn *LocalConnection) sendQuery(code int, payload interface{}) {
-	select {
-	case conn.queryChan <- &ConnectionInteraction{
-		Interaction: Interaction{code: code},
-		payload:     payload}:
-	case <-conn.finished:
-	}
-}
+// NB: The conn.* fields are only written by the connection actor
+// process, which is the caller of the ConnectionAction funs. Hence we
+// do not need locks for reading, and only need write locks for fields
+// read by other processes.
 
 // Async
 func (conn *LocalConnection) Shutdown(err error) {
-	// Run on its own goroutine in case the channel is backed up
-	go conn.sendQuery(CShutdown, err)
+	if err == nil {
+		conn.actionChan <- nil
+	} else {
+		// Run on its own goroutine in case the channel is backed up
+		go func() { conn.sendAction(func() error { return err }) }()
+	}
 }
 
 // Async
@@ -216,18 +205,76 @@ func (conn *LocalConnection) ReceivedHeartbeat(remoteUDPAddr *net.UDPAddr, connU
 	if remoteUDPAddr == nil || connUID != conn.uid {
 		return
 	}
-	conn.sendQuery(CReceivedHeartbeat, remoteUDPAddr)
+	conn.sendAction(func() error {
+		oldRemoteUDPAddr := conn.remoteUDPAddr
+		old := conn.receivedHeartbeat
+		conn.Lock()
+		conn.remoteUDPAddr = remoteUDPAddr
+		conn.receivedHeartbeat = true
+		conn.Unlock()
+		conn.heartbeatTimeout.Reset(HeartbeatTimeout)
+		if !old {
+			if err := conn.sendSimpleProtocolMsg(ProtocolConnectionEstablished); err != nil {
+				return err
+			}
+		}
+		if oldRemoteUDPAddr == nil {
+			return conn.sendFastHeartbeats()
+		} else if oldRemoteUDPAddr.String() != remoteUDPAddr.String() {
+			log.Println("Peer", conn.remote.Name, "moved from", old, "to", remoteUDPAddr)
+		}
+		return nil
+	})
 }
 
 // Async
 func (conn *LocalConnection) SetEstablished() {
-	conn.sendQuery(CSetEstablished, nil)
+	conn.sendAction(func() error {
+		stopTicker(conn.heartbeat)
+		old := conn.established
+		conn.Lock()
+		conn.established = true
+		conn.Unlock()
+		if old {
+			return nil
+		}
+		conn.Router.Ourself.ConnectionEstablished(conn)
+		if err := conn.ensureForwarders(); err != nil {
+			return err
+		}
+		// Send a large frame down the DF channel in order to prompt
+		// PMTU discovery to start.
+		conn.Forward(true, &ForwardedFrame{
+			srcPeer: conn.local,
+			dstPeer: conn.remote,
+			frame:   PMTUDiscovery},
+			nil)
+		conn.heartbeat = time.NewTicker(SlowHeartbeat)
+		conn.fragTest = time.NewTicker(FragTestInterval)
+		// avoid initial waits for timers to fire
+		conn.Forward(true, conn.heartbeatFrame, nil)
+		conn.setStackFrag(false)
+		if err := conn.sendSimpleProtocolMsg(ProtocolStartFragmentationTest); err != nil {
+			return err
+		}
+		return nil
+	})
+}
+
+// Send an actor request to the actorLoop, but don't block if
+// actorLoop has exited - see http://blog.golang.org/pipelines for
+// pattern
+func (conn *LocalConnection) sendAction(action ConnectionAction) {
+	select {
+	case conn.actionChan <- action:
+	case <-conn.finished:
+	}
 }
 
 // ACTOR server
 
-func (conn *LocalConnection) run(queryChan <-chan *ConnectionInteraction, finished chan<- struct{}, acceptNewPeer bool) {
-	defer conn.handleShutdown()
+func (conn *LocalConnection) run(actionChan <-chan ConnectionAction, finished chan<- struct{}, acceptNewPeer bool) {
+	defer conn.shutdown()
 	defer close(finished)
 
 	tcpConn := conn.TCPConn
@@ -268,7 +315,7 @@ func (conn *LocalConnection) run(queryChan <-chan *ConnectionInteraction, finish
 
 	conn.heartbeatTimeout = time.NewTimer(HeartbeatTimeout)
 
-	if err := conn.queryLoop(queryChan); err != nil {
+	if err := conn.actorLoop(actionChan); err != nil {
 		conn.Log("connection shutting down due to error:", err)
 	} else {
 		conn.Log("connection shutting down")
@@ -286,23 +333,14 @@ func (conn *LocalConnection) run(queryChan <-chan *ConnectionInteraction, finish
 // AddConnection, which is *synchronous*, before entering the tcp
 // receive loop.
 
-func (conn *LocalConnection) queryLoop(queryChan <-chan *ConnectionInteraction) (err error) {
-	terminate := false
-	for !terminate && err == nil {
+func (conn *LocalConnection) actorLoop(actionChan <-chan ConnectionAction) (err error) {
+	for err == nil {
 		select {
-		case query, ok := <-queryChan:
-			if !ok {
+		case action := <-actionChan:
+			if action == nil {
 				break
 			}
-			switch query.code {
-			case CShutdown:
-				err = query.payload.(error)
-				terminate = true
-			case CReceivedHeartbeat:
-				err = conn.handleReceivedHeartbeat(query.payload.(*net.UDPAddr))
-			case CSetEstablished:
-				err = conn.handleSetEstablished()
-			}
+			err = action()
 		case <-conn.heartbeatTimeout.C:
 			err = fmt.Errorf("timed out waiting for UDP heartbeat")
 		case <-tickerChan(conn.heartbeat):
@@ -315,66 +353,7 @@ func (conn *LocalConnection) queryLoop(queryChan <-chan *ConnectionInteraction)
 	return
 }
 
-// Handlers
-//
-// NB: The conn.* fields are only written by the connection actor
-// process, which is the caller of the handlers. Hence we do not need
-// locks for reading, and only need write locks for fields read by
-// other processes.
-
-func (conn *LocalConnection) handleReceivedHeartbeat(remoteUDPAddr *net.UDPAddr) error {
-	oldRemoteUDPAddr := conn.remoteUDPAddr
-	old := conn.receivedHeartbeat
-	conn.Lock()
-	conn.remoteUDPAddr = remoteUDPAddr
-	conn.receivedHeartbeat = true
-	conn.Unlock()
-	conn.heartbeatTimeout.Reset(HeartbeatTimeout)
-	if !old {
-		if err := conn.sendSimpleProtocolMsg(ProtocolConnectionEstablished); err != nil {
-			return err
-		}
-	}
-	if oldRemoteUDPAddr == nil {
-		return conn.sendFastHeartbeats()
-	} else if oldRemoteUDPAddr.String() != remoteUDPAddr.String() {
-		log.Println("Peer", conn.remote.Name, "moved from", old, "to", remoteUDPAddr)
-	}
-	return nil
-}
-
-func (conn *LocalConnection) handleSetEstablished() error {
-	stopTicker(conn.heartbeat)
-	old := conn.established
-	conn.Lock()
-	conn.established = true
-	conn.Unlock()
-	if old {
-		return nil
-	}
-	conn.Router.Ourself.ConnectionEstablished(conn)
-	if err := conn.ensureForwarders(); err != nil {
-		return err
-	}
-	// Send a large frame down the DF channel in order to prompt
-	// PMTU discovery to start.
-	conn.Forward(true, &ForwardedFrame{
-		srcPeer: conn.local,
-		dstPeer: conn.remote,
-		frame:   PMTUDiscovery},
-		nil)
-	conn.heartbeat = time.NewTicker(SlowHeartbeat)
-	conn.fragTest = time.NewTicker(FragTestInterval)
-	// avoid initial waits for timers to fire
-	conn.Forward(true, conn.heartbeatFrame, nil)
-	conn.setStackFrag(false)
-	if err := conn.sendSimpleProtocolMsg(ProtocolStartFragmentationTest); err != nil {
-		return err
-	}
-	return nil
-}
-
-func (conn *LocalConnection) handleShutdown() {
+func (conn *LocalConnection) shutdown() {
 	if conn.TCPConn != nil {
 		checkWarn(conn.TCPConn.Close())
 	}

router/connection_maker.go

@@ -14,19 +14,12 @@ const (
 	MaxInterval     = 10 * time.Minute
 )
 
-const (
-	CMInitiate = iota
-	CMTerminated
-	CMRefresh
-	CMStatus
-)
-
 type ConnectionMaker struct {
 	ourself        *LocalPeer
 	peers          *Peers
 	targets        map[string]*Target
 	cmdLineAddress map[string]bool
-	queryChan      chan<- *ConnectionMakerInteraction
+	actionChan     chan<- ConnectionMakerAction
 }
 
 // Information about an address where we may find a peer
@@ -36,10 +29,7 @@ type Target struct {
 	tryInterval time.Duration // backoff time on next failure
 }
 
-type ConnectionMakerInteraction struct {
-	Interaction
-	address string
-}
+type ConnectionMakerAction func() bool
 
 func NewConnectionMaker(ourself *LocalPeer, peers *Peers) *ConnectionMaker {
 	return &ConnectionMaker{
@@ -50,61 +40,59 @@ func NewConnectionMaker(ourself *LocalPeer, peers *Peers) *ConnectionMaker {
 }
 
 func (cm *ConnectionMaker) Start() {
-	queryChan := make(chan *ConnectionMakerInteraction, ChannelSize)
-	cm.queryChan = queryChan
-	go cm.queryLoop(queryChan)
+	actionChan := make(chan ConnectionMakerAction, ChannelSize)
+	cm.actionChan = actionChan
+	go cm.queryLoop(actionChan)
 }
 
 func (cm *ConnectionMaker) InitiateConnection(address string) {
-	cm.queryChan <- &ConnectionMakerInteraction{
-		Interaction: Interaction{code: CMInitiate},
-		address:     address}
+	cm.actionChan <- func() bool {
+		cm.cmdLineAddress[NormalisePeerAddr(address)] = true
+		return true
+	}
 }
 
 func (cm *ConnectionMaker) ConnectionTerminated(address string) {
-	cm.queryChan <- &ConnectionMakerInteraction{
-		Interaction: Interaction{code: CMTerminated},
-		address:     address}
+	cm.actionChan <- func() bool {
+		if target, found := cm.targets[address]; found {
+			target.attempting = false
+			target.tryAfter, target.tryInterval = tryAfter(target.tryInterval)
+		}
+		return true
+	}
 }
 
 func (cm *ConnectionMaker) Refresh() {
-	cm.queryChan <- &ConnectionMakerInteraction{
-		Interaction: Interaction{code: CMRefresh}}
+	cm.actionChan <- func() bool { return true }
 }
 
 func (cm *ConnectionMaker) String() string {
-	resultChan := make(chan interface{}, 0)
-	cm.queryChan <- &ConnectionMakerInteraction{
-		Interaction: Interaction{code: CMStatus, resultChan: resultChan}}
-	result := <-resultChan
-	return result.(string)
+	resultChan := make(chan string, 0)
+	cm.actionChan <- func() bool {
+		var buf bytes.Buffer
+		for address, target := range cm.targets {
+			var fmtStr string
+			if target.attempting {
+				fmtStr = "%s (trying since %v)\n"
+			} else {
+				fmtStr = "%s (next try at %v)\n"
+			}
+			buf.WriteString(fmt.Sprintf(fmtStr, address, target.tryAfter))
+		}
+		resultChan <- buf.String()
+		return false
+	}
+	return <-resultChan
 }
 
-func (cm *ConnectionMaker) queryLoop(queryChan <-chan *ConnectionMakerInteraction) {
+func (cm *ConnectionMaker) queryLoop(actionChan <-chan ConnectionMakerAction) {
 	timer := time.NewTimer(MaxDuration)
 	run := func() { timer.Reset(cm.checkStateAndAttemptConnections()) }
 	for {
 		select {
-		case query, ok := <-queryChan:
-			if !ok {
-				return
-			}
-			switch query.code {
-			case CMInitiate:
-				cm.cmdLineAddress[NormalisePeerAddr(query.address)] = true
+		case action := <-actionChan:
+			if action() {
 				run()
-			case CMTerminated:
-				if target, found := cm.targets[query.address]; found {
-					target.attempting = false
-					target.tryAfter, target.tryInterval = tryAfter(target.tryInterval)
-				}
-				run()
-			case CMRefresh:
-				run()
-			case CMStatus:
-				query.resultChan <- cm.status()
-			default:
-				log.Fatal("Unexpected connection maker query:", query)
 			}
 		case <-timer.C:
 			run()
@@ -187,20 +175,6 @@ func (cm *ConnectionMaker) addTarget(address string) {
 	}
 }
 
-func (cm *ConnectionMaker) status() string {
-	var buf bytes.Buffer
-	for address, target := range cm.targets {
-		var fmtStr string
-		if target.attempting {
-			fmtStr = "%s (trying since %v)\n"
-		} else {
-			fmtStr = "%s (next try at %v)\n"
-		}
-		buf.WriteString(fmt.Sprintf(fmtStr, address, target.tryAfter))
-	}
-	return buf.String()
-}
-
 func (cm *ConnectionMaker) attemptConnection(address string, acceptNewPeer bool) {
 	log.Printf("->[%s] attempting connection\n", address)
 	if err := cm.ourself.CreateConnection(address, acceptNewPeer); err != nil {

router/gossip_test.go

@@ -20,8 +20,8 @@ type mockChannelConnection struct {
 // channels when Router.OnGossip() calls async methods.
 func NewTestRouter(name PeerName) *Router {
 	router := NewRouter(nil, name, "", nil, 10, 1024, nil)
-	router.ConnectionMaker.queryChan = make(chan *ConnectionMakerInteraction, ChannelSize)
-	router.Routes.queryChan = make(chan *Interaction, ChannelSize)
+	router.ConnectionMaker.actionChan = make(chan ConnectionMakerAction, ChannelSize)
+	router.Routes.actionChan = make(chan RoutesAction, ChannelSize)
 	return router
 }
 

router/local_peer.go

@@ -9,23 +9,20 @@ import (
 
 type LocalPeer struct {
 	*Peer
-	Router    *Router
-	queryChan chan<- *PeerInteraction
+	Router     *Router
+	actionChan chan<- LocalPeerAction
 }
 
-type PeerInteraction struct {
-	Interaction
-	payload interface{}
-}
+type LocalPeerAction func()
 
 func NewLocalPeer(name PeerName, nickName string, router *Router) *LocalPeer {
 	return &LocalPeer{Peer: NewPeer(name, nickName, 0, 0), Router: router}
 }
 
 func (peer *LocalPeer) Start() {
-	queryChan := make(chan *PeerInteraction, ChannelSize)
-	peer.queryChan = queryChan
-	go peer.queryLoop(queryChan)
+	actionChan := make(chan LocalPeerAction, ChannelSize)
+	peer.actionChan = actionChan
+	go peer.actorLoop(actionChan)
 }
 
 func (peer *LocalPeer) Forward(dstPeer *Peer, df bool, frame []byte, dec *EthernetDecoder) error {
@@ -128,48 +125,38 @@ const (
 // Sync.
 func (peer *LocalPeer) AddConnection(conn *LocalConnection) {
 	resultChan := make(chan interface{})
-	peer.queryChan <- &PeerInteraction{
-		Interaction: Interaction{code: PAddConnection, resultChan: resultChan},
-		payload:     conn}
+	peer.actionChan <- func() {
+		peer.handleAddConnection(conn)
+		resultChan <- nil
+	}
 	<-resultChan
 }
 
 // Async.
 func (peer *LocalPeer) ConnectionEstablished(conn *LocalConnection) {
-	peer.queryChan <- &PeerInteraction{
-		Interaction: Interaction{code: PConnectionEstablished},
-		payload:     conn}
+	peer.actionChan <- func() {
+		peer.handleConnectionEstablished(conn)
+	}
 }
 
 // Sync.
 func (peer *LocalPeer) DeleteConnection(conn *LocalConnection) {
 	resultChan := make(chan interface{})
-	peer.queryChan <- &PeerInteraction{
-		Interaction: Interaction{code: PDeleteConnection, resultChan: resultChan},
-		payload:     conn}
+	peer.actionChan <- func() {
+		peer.handleDeleteConnection(conn)
+		resultChan <- nil
+	}
 	<-resultChan
 }
 
 // ACTOR server
 
-func (peer *LocalPeer) queryLoop(queryChan <-chan *PeerInteraction) {
+func (peer *LocalPeer) actorLoop(actionChan <-chan LocalPeerAction) {
 	gossipTimer := time.Tick(GossipInterval)
 	for {
 		select {
-		case query, ok := <-queryChan:
-			if !ok {
-				return
-			}
-			switch query.code {
-			case PAddConnection:
-				peer.handleAddConnection(query.payload.(*LocalConnection))
-				query.resultChan <- nil
-			case PConnectionEstablished:
-				peer.handleConnectionEstablished(query.payload.(*LocalConnection))
-			case PDeleteConnection:
-				peer.handleDeleteConnection(query.payload.(*LocalConnection))
-				query.resultChan <- nil
-			}
+		case action := <-actionChan:
+			action()
 		case <-gossipTimer:
 			peer.Router.SendAllGossip()
 		}

router/routes.go

@@ -3,17 +3,16 @@ package router
 import (
 	"bytes"
 	"fmt"
-	"log"
 	"sync"
 )
 
 type Routes struct {
 	sync.RWMutex
-	ourself   *Peer
-	peers     *Peers
-	unicast   map[PeerName]PeerName
-	broadcast map[PeerName][]PeerName
-	queryChan chan<- *Interaction
+	ourself    *Peer
+	peers      *Peers
+	unicast    map[PeerName]PeerName
+	broadcast  map[PeerName][]PeerName
+	actionChan chan<- RoutesAction
 }
 
 func NewRoutes(ourself *Peer, peers *Peers) *Routes {
@@ -28,9 +27,9 @@ func NewRoutes(ourself *Peer, peers *Peers) *Routes {
 }
 
 func (routes *Routes) Start() {
-	queryChan := make(chan *Interaction, ChannelSize)
-	routes.queryChan = queryChan
-	go routes.queryLoop(queryChan)
+	actionChan := make(chan RoutesAction, ChannelSize)
+	routes.actionChan = actionChan
+	go routes.actorLoop(actionChan)
 }
 
 func (routes *Routes) Unicast(name PeerName) (PeerName, bool) {
@@ -67,34 +66,25 @@ func (routes *Routes) String() string {
 
 // ACTOR client API
 
-const (
-	RRecalculate = iota
-)
+type RoutesAction func()
 
 // Async.
 func (routes *Routes) Recalculate() {
-	routes.queryChan <- &Interaction{code: RRecalculate}
+	routes.actionChan <- func() {
+		unicast := routes.calculateUnicast()
+		broadcast := routes.calculateBroadcast()
+		routes.Lock()
+		routes.unicast = unicast
+		routes.broadcast = broadcast
+		routes.Unlock()
+	}
 }
 
 // ACTOR server
 
-func (routes *Routes) queryLoop(queryChan <-chan *Interaction) {
+func (routes *Routes) actorLoop(actionChan <-chan RoutesAction) {
 	for {
-		query, ok := <-queryChan
-		if !ok {
-			return
-		}
-		switch query.code {
-		case RRecalculate:
-			unicast := routes.calculateUnicast()
-			broadcast := routes.calculateBroadcast()
-			routes.Lock()
-			routes.unicast = unicast
-			routes.broadcast = broadcast
-			routes.Unlock()
-		default:
-			log.Fatal("Unexpected routes query:", query)
-		}
+		(<-actionChan)()
 	}
 }
 

router/utils.go

@@ -10,11 +10,6 @@ import (
 	"net"
 )
 
-type Interaction struct {
-	code       int
-	resultChan chan<- interface{}
-}
-
 func checkFatal(e error) {
 	if e != nil {
 		log.Fatal(e)