consul: refactor into more files

This commit is contained in:
Armon Dadgar 2013-12-06 16:54:33 -08:00
parent 310eb2f506
commit 41b3ca7da9
3 changed files with 71 additions and 65 deletions

46
consul/rpc.go Normal file
View File

@ -0,0 +1,46 @@
package consul
import (
"github.com/ugorji/go/codec"
"net"
)
// listen is used to listen for incoming RPC connections
func (s *Server) listen() {
for {
// Accept a connection
conn, err := s.rpcListener.Accept()
if err != nil {
if s.shutdown {
return
}
s.logger.Printf("[ERR] Failed to accept RPC conn: %v", err)
continue
}
// Track this client
s.rpcClientLock.Lock()
s.rpcClients[conn] = struct{}{}
s.rpcClientLock.Unlock()
go s.handleConn(conn)
}
}
// handleConn is used to service a single RPC connection
func (s *Server) handleConn(conn net.Conn) {
defer func() {
conn.Close()
s.rpcClientLock.Lock()
delete(s.rpcClients, conn)
s.rpcClientLock.Unlock()
}()
rpcCodec := codec.GoRpc.ServerCodec(conn, &codec.MsgpackHandle{})
for !s.shutdown {
if err := s.rpcServer.ServeRequest(rpcCodec); err != nil {
s.logger.Printf("[ERR] RPC error: %v (%v)", err, conn)
return
}
}
}

25
consul/serf.go Normal file
View File

@ -0,0 +1,25 @@
package connsul
// lanEventHandler is used to handle events from the lan Serf cluster
func (s *Server) lanEventHandler() {
for {
select {
case e := <-s.eventChLAN:
s.logger.Printf("[INFO] LAN Event: %v", e)
case <-s.shutdownCh:
return
}
}
}
// wanEventHandler is used to handle events from the wan Serf cluster
func (s *Server) wanEventHandler() {
for {
select {
case e := <-s.eventChWAN:
s.logger.Printf("[INFO] WAN Event: %v", e)
case <-s.shutdownCh:
return
}
}
}

View File

@ -4,7 +4,6 @@ import (
"fmt"
"github.com/hashicorp/raft"
"github.com/hashicorp/serf/serf"
"github.com/ugorji/go/codec"
"log"
"net"
"net/rpc"
@ -252,67 +251,3 @@ func (s *Server) Shutdown() error {
return nil
}
// lanEventHandler is used to handle events from the lan Serf cluster
func (s *Server) lanEventHandler() {
for {
select {
case e := <-s.eventChLAN:
s.logger.Printf("[INFO] LAN Event: %v", e)
case <-s.shutdownCh:
return
}
}
}
// wanEventHandler is used to handle events from the wan Serf cluster
func (s *Server) wanEventHandler() {
for {
select {
case e := <-s.eventChWAN:
s.logger.Printf("[INFO] WAN Event: %v", e)
case <-s.shutdownCh:
return
}
}
}
// listen is used to listen for incoming RPC connections
func (s *Server) listen() {
for {
// Accept a connection
conn, err := s.rpcListener.Accept()
if err != nil {
if s.shutdown {
return
}
s.logger.Printf("[ERR] Failed to accept RPC conn: %v", err)
continue
}
// Track this client
s.rpcClientLock.Lock()
s.rpcClients[conn] = struct{}{}
s.rpcClientLock.Unlock()
go s.handleConn(conn)
}
}
// handleConn is used to service a single RPC connection
func (s *Server) handleConn(conn net.Conn) {
defer func() {
conn.Close()
s.rpcClientLock.Lock()
delete(s.rpcClients, conn)
s.rpcClientLock.Unlock()
}()
rpcCodec := codec.GoRpc.ServerCodec(conn, &codec.MsgpackHandle{})
for !s.shutdown {
if err := s.rpcServer.ServeRequest(rpcCodec); err != nil {
s.logger.Printf("[ERR] RPC error: %v (%v)", err, conn)
return
}
}
}