func (s *Server) setQueryMeta(m *structs.QueryMeta) {
	if s.IsLeader() {
		m.LastContact = 0
		m.KnownLeader = true
	} else {
		m.LastContact = time.Now().Sub(s.raft.LastContact())
		m.KnownLeader = (s.raft.Leader() != nil)
	}
}
setQueryMeta is used to populate the QueryMeta data for an RPC call
package consul
import (
	"crypto/tls"
	"fmt"
	"io"
	"math/rand"
	"net"
	"strings"
	"time"
	"github.com/socketplane/socketplane/Godeps/_workspace/src/github.com/armon/go-metrics"
	"github.com/socketplane/socketplane/Godeps/_workspace/src/github.com/hashicorp/consul/consul/structs"
	"github.com/socketplane/socketplane/Godeps/_workspace/src/github.com/hashicorp/go-msgpack/codec"
	"github.com/socketplane/socketplane/Godeps/_workspace/src/github.com/hashicorp/yamux"
	"github.com/socketplane/socketplane/Godeps/_workspace/src/github.com/inconshreveable/muxado"
)
type RPCType byte
const (
	rpcConsul RPCType = iota
	rpcRaft
	rpcMultiplex
	rpcTLS
	rpcMultiplexV2
)
const (
	maxQueryTime = 600 * time.Second
	raftWarnSize = 1024 * 1024
	enqueueLimit = 30 * time.Second
)
func (s *Server) listen() {
	for {
		conn, err := s.rpcListener.Accept()
		if err != nil {
			if s.shutdown {
				return
			}
			s.logger.Printf("[ERR] consul.rpc: failed to accept RPC conn: %v", err)
			continue
		}
		go s.handleConn(conn, false)
		metrics.IncrCounter([]string{"consul", "rpc", "accept_conn"}, 1)
	}
}
func (s *Server) handleConn(conn net.Conn, isTLS bool) {
	buf := make([]byte, 1)
	if _, err := conn.Read(buf); err != nil {
		if err != io.EOF {
			s.logger.Printf("[ERR] consul.rpc: failed to read byte: %v", err)
		}
		conn.Close()
		return
	}
	if s.config.VerifyIncoming && !isTLS && RPCType(buf[0]) != rpcTLS {
		s.logger.Printf("[WARN] consul.rpc: Non-TLS connection attempted with VerifyIncoming set")
		conn.Close()
		return
	}
	switch RPCType(buf[0]) {
	case rpcConsul:
		s.handleConsulConn(conn)
	case rpcRaft:
		metrics.IncrCounter([]string{"consul", "rpc", "raft_handoff"}, 1)
		s.raftLayer.Handoff(conn)
	case rpcMultiplex:
		s.handleMultiplex(conn)
	case rpcTLS:
		if s.rpcTLS == nil {
			s.logger.Printf("[WARN] consul.rpc: TLS connection attempted, server not configured for TLS")
			conn.Close()
			return
		}
		conn = tls.Server(conn, s.rpcTLS)
		s.handleConn(conn, true)
	case rpcMultiplexV2:
		s.handleMultiplexV2(conn)
	default:
		s.logger.Printf("[ERR] consul.rpc: unrecognized RPC byte: %v", buf[0])
		conn.Close()
		return
	}
}
func (s *Server) handleMultiplex(conn net.Conn) {
	defer conn.Close()
	server := muxado.Server(conn)
	for {
		sub, err := server.Accept()
		if err != nil {
			if !strings.Contains(err.Error(), "closed") {
				s.logger.Printf("[ERR] consul.rpc: multiplex conn accept failed: %v", err)
			}
			return
		}
		go s.handleConsulConn(sub)
	}
}
func (s *Server) handleMultiplexV2(conn net.Conn) {
	defer conn.Close()
	conf := yamux.DefaultConfig()
	conf.LogOutput = s.config.LogOutput
	server, _ := yamux.Server(conn, conf)
	for {
		sub, err := server.Accept()
		if err != nil {
			if err != io.EOF {
				s.logger.Printf("[ERR] consul.rpc: multiplex conn accept failed: %v", err)
			}
			return
		}
		go s.handleConsulConn(sub)
	}
}
func (s *Server) handleConsulConn(conn net.Conn) {
	defer conn.Close()
	rpcCodec := codec.GoRpc.ServerCodec(conn, msgpackHandle)
	for {
		select {
		case <-s.shutdownCh:
			return
		default:
		}
		if err := s.rpcServer.ServeRequest(rpcCodec); err != nil {
			if err != io.EOF && !strings.Contains(err.Error(), "closed") {
				s.logger.Printf("[ERR] consul.rpc: RPC error: %v (%v)", err, conn)
			}
			return
		}
	}
}
func (s *Server) forward(method string, info structs.RPCInfo, args interface{}, reply interface{}) (bool, error) {
	dc := info.RequestDatacenter()
	if dc != s.config.Datacenter {
		err := s.forwardDC(method, dc, args, reply)
		return true, err
	}
	if info.IsRead() && info.AllowStaleRead() {
		return false, nil
	}
	if !s.IsLeader() {
		err := s.forwardLeader(method, args, reply)
		return true, err
	}
	return false, nil
}
func (s *Server) forwardLeader(method string, args interface{}, reply interface{}) error {
	leader := s.raft.Leader()
	if leader == nil {
		return structs.ErrNoLeader
	}
	s.localLock.RLock()
	server := s.localConsuls[leader.String()]
	s.localLock.RUnlock()
	if server == nil {
		return structs.ErrNoLeader
	}
	return s.connPool.RPC(server.Addr, server.Version, method, args, reply)
}
func (s *Server) forwardDC(method, dc string, args interface{}, reply interface{}) error {
	s.remoteLock.RLock()
	servers := s.remoteConsuls[dc]
	if len(servers) == 0 {
		s.remoteLock.RUnlock()
		s.logger.Printf("[WARN] consul.rpc: RPC request for DC '%s', no path found", dc)
		return structs.ErrNoDCPath
	}
	offset := rand.Int31() % int32(len(servers))
	server := servers[offset]
	s.remoteLock.RUnlock()
	metrics.IncrCounter([]string{"consul", "rpc", "cross-dc", dc}, 1)
	return s.connPool.RPC(server.Addr, server.Version, method, args, reply)
}
func (s *Server) globalRPC(method string, args interface{},
	reply structs.CompoundResponse) error {
	errorCh := make(chan error)
	respCh := make(chan interface{})
	for dc, _ := range s.remoteConsuls {
		go func(dc string) {
			rr := reply.New()
			if err := s.forwardDC(method, dc, args, &rr); err != nil {
				errorCh <- err
				return
			}
			respCh <- rr
		}(dc)
	}
	replies, total := 0, len(s.remoteConsuls)
	for replies < total {
		select {
		case err := <-errorCh:
			return err
		case rr := <-respCh:
			reply.Add(rr)
			replies++
		}
	}
	return nil
}
func (s *Server) raftApply(t structs.MessageType, msg interface{}) (interface{}, error) {
	buf, err := structs.Encode(t, msg)
	if err != nil {
		return nil, fmt.Errorf("Failed to encode request: %v", err)
	}
	if n := len(buf); n > raftWarnSize {
		s.logger.Printf("[WARN] consul: Attempting to apply large raft entry (%d bytes)", n)
	}
	future := s.raft.Apply(buf, enqueueLimit)
	if err := future.Error(); err != nil {
		return nil, err
	}
	return future.Response(), nil
}
func (s *Server) blockingRPC(b *structs.QueryOptions, m *structs.QueryMeta,
	tables MDBTables, run func() error) error {
	opts := blockingRPCOptions{
		queryOpts: b,
		queryMeta: m,
		tables:    tables,
		run:       run,
	}
	return s.blockingRPCOpt(&opts)
}
type blockingRPCOptions struct {
	queryOpts *structs.QueryOptions
	queryMeta *structs.QueryMeta
	tables    MDBTables
	kvWatch   bool
	kvPrefix  string
	run       func() error
}
func (s *Server) blockingRPCOpt(opts *blockingRPCOptions) error {
	var timeout <-chan time.Time
	var notifyCh chan struct{}
	if opts.queryOpts.MinQueryIndex == 0 {
		goto RUN_QUERY
	}
	if len(opts.tables) == 0 && !opts.kvWatch {
		panic("no tables to block on")
	}
	if opts.queryOpts.MaxQueryTime > maxQueryTime {
		opts.queryOpts.MaxQueryTime = maxQueryTime
	}
	if opts.queryOpts.MinQueryIndex > 0 && opts.queryOpts.MaxQueryTime == 0 {
		opts.queryOpts.MaxQueryTime = maxQueryTime
	}
	if opts.queryOpts.MaxQueryTime > 0 {
		timeout = time.After(opts.queryOpts.MaxQueryTime)
	}
SETUP_NOTIFY:
	if opts.queryOpts.MinQueryIndex > 0 {
		notifyCh = make(chan struct{}, 1)
		state := s.fsm.State()
		state.Watch(opts.tables, notifyCh)
		if opts.kvWatch {
			state.WatchKV(opts.kvPrefix, notifyCh)
		}
	}
RUN_QUERY:
	s.setQueryMeta(opts.queryMeta)
	if opts.queryOpts.RequireConsistent {
		if err := s.consistentRead(); err != nil {
			return err
		}
	}
	err := opts.run()
	if err == nil && opts.queryMeta.Index > 0 && opts.queryMeta.Index <= opts.queryOpts.MinQueryIndex {
		select {
		case <-notifyCh:
			goto SETUP_NOTIFY
		case <-timeout:
		}
	}
	return err
}