open-consul/consul/rpc.go

130 lines
3.1 KiB
Go
Raw Normal View History

2013-12-07 00:54:33 +00:00
package consul
import (
"fmt"
"github.com/hashicorp/consul/rpc"
2013-12-07 00:54:33 +00:00
"github.com/ugorji/go/codec"
"net"
)
type RPCType byte
const (
rpcConsul RPCType = iota
rpcRaft
)
2013-12-07 00:54:33 +00:00
// 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 determine if this is a Raft or
// Consul type RPC connection and invoke the correct handler
2013-12-07 00:54:33 +00:00
func (s *Server) handleConn(conn net.Conn) {
// Read a single byte
buf := make([]byte, 1)
if _, err := conn.Read(buf); err != nil {
s.logger.Printf("[ERR] Failed to read byte: %v", err)
conn.Close()
return
}
// Switch on the byte
switch RPCType(buf[0]) {
case rpcConsul:
s.handleConsulConn(conn)
case rpcRaft:
s.raftLayer.Handoff(conn)
default:
s.logger.Printf("[ERR] Unrecognized RPC byte: %v", buf[0])
conn.Close()
return
}
}
// handleConsulConn is used to service a single Consul RPC connection
func (s *Server) handleConsulConn(conn net.Conn) {
2013-12-07 00:54:33 +00:00
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
}
}
}
// forward is used to forward to a remote DC or to forward to the local leader
// Returns a bool of if forwarding was performed, as well as any error
func (s *Server) forward(method, dc string, args interface{}, reply interface{}) (bool, error) {
// Handle DC forwarding
if dc != s.config.Datacenter {
err := s.forwardDC(method, dc, args, reply)
return true, err
}
// Handle leader forwarding
if !s.IsLeader() {
err := s.forwardLeader(method, args, reply)
return true, err
}
return false, nil
}
// forwardLeader is used to forward an RPC call to the leader, or fail if no leader
func (s *Server) forwardLeader(method string, args interface{}, reply interface{}) error {
leader := s.raft.Leader()
if leader == nil {
return rpc.ErrNoLeader
}
return s.connPool.RPC(leader, method, args, reply)
}
// forwardDC is used to forward an RPC call to a remote DC, or fail if no servers
func (s *Server) forwardDC(method, dc string, args interface{}, reply interface{}) error {
// TODO: Fix
return fmt.Errorf("DC forwarding not supported")
}
// raftApply is used to encode a message, run it through raft, and return
// the FSM response along with any errors
func (s *Server) raftApply(t rpc.MessageType, msg interface{}) (interface{}, error) {
buf, err := rpc.Encode(t, msg)
if err != nil {
return nil, fmt.Errorf("Failed to encode request: %v", err)
}
future := s.raft.Apply(buf, 0)
if err := future.Error(); err != nil {
return nil, err
}
return future.Response(), nil
}