func (c *RPCClient) handleSeq(seq uint64, handler seqHandler) {
	c.dispatchLock.Lock()
	defer c.dispatchLock.Unlock()
	c.dispatch[seq] = handler
}
handleSeq is used to setup a handlerto wait on a response for
a given sequence number.
package agent
import (
	"bufio"
	"fmt"
	"github.com/socketplane/socketplane/Godeps/_workspace/src/github.com/hashicorp/go-msgpack/codec"
	"github.com/socketplane/socketplane/Godeps/_workspace/src/github.com/hashicorp/logutils"
	"log"
	"net"
	"os"
	"strings"
	"sync"
	"sync/atomic"
)
var (
	clientClosed = fmt.Errorf("client closed")
)
type seqCallback struct {
	handler func(*responseHeader)
}
func (sc *seqCallback) Handle(resp *responseHeader) {
	sc.handler(resp)
}
func (sc *seqCallback) Cleanup() {}
type seqHandler interface {
	Handle(*responseHeader)
	Cleanup()
}
type RPCClient struct {
	seq uint64
	conn      net.Conn
	reader    *bufio.Reader
	writer    *bufio.Writer
	dec       *codec.Decoder
	enc       *codec.Encoder
	writeLock sync.Mutex
	dispatch     map[uint64]seqHandler
	dispatchLock sync.Mutex
	shutdown     bool
	shutdownCh   chan struct{}
	shutdownLock sync.Mutex
}
func (c *RPCClient) send(header *requestHeader, obj interface{}) error {
	c.writeLock.Lock()
	defer c.writeLock.Unlock()
	if c.shutdown {
		return clientClosed
	}
	if err := c.enc.Encode(header); err != nil {
		return err
	}
	if obj != nil {
		if err := c.enc.Encode(obj); err != nil {
			return err
		}
	}
	if err := c.writer.Flush(); err != nil {
		return err
	}
	return nil
}
func NewRPCClient(addr string) (*RPCClient, error) {
	var conn net.Conn
	var err error
	if envAddr := os.Getenv("CONSUL_RPC_ADDR"); envAddr != "" {
		addr = envAddr
	}
	mode := "tcp"
	if strings.HasPrefix(addr, "/") {
		mode = "unix"
	}
	if conn, err = net.Dial(mode, addr); err != nil {
		return nil, err
	}
	client := &RPCClient{
		seq:        0,
		conn:       conn,
		reader:     bufio.NewReader(conn),
		writer:     bufio.NewWriter(conn),
		dispatch:   make(map[uint64]seqHandler),
		shutdownCh: make(chan struct{}),
	}
	client.dec = codec.NewDecoder(client.reader, msgpackHandle)
	client.enc = codec.NewEncoder(client.writer, msgpackHandle)
	go client.listen()
	if err := client.handshake(); err != nil {
		client.Close()
		return nil, err
	}
	return client, err
}
type StreamHandle uint64
func (c *RPCClient) Close() error {
	c.shutdownLock.Lock()
	defer c.shutdownLock.Unlock()
	if !c.shutdown {
		c.shutdown = true
		close(c.shutdownCh)
		c.deregisterAll()
		return c.conn.Close()
	}
	return nil
}
func (c *RPCClient) ForceLeave(node string) error {
	header := requestHeader{
		Command: forceLeaveCommand,
		Seq:     c.getSeq(),
	}
	req := forceLeaveRequest{
		Node: node,
	}
	return c.genericRPC(&header, &req, nil)
}
func (c *RPCClient) Join(addrs []string, wan bool) (int, error) {
	header := requestHeader{
		Command: joinCommand,
		Seq:     c.getSeq(),
	}
	req := joinRequest{
		Existing: addrs,
		WAN:      wan,
	}
	var resp joinResponse
	err := c.genericRPC(&header, &req, &resp)
	return int(resp.Num), err
}
func (c *RPCClient) LANMembers() ([]Member, error) {
	header := requestHeader{
		Command: membersLANCommand,
		Seq:     c.getSeq(),
	}
	var resp membersResponse
	err := c.genericRPC(&header, nil, &resp)
	return resp.Members, err
}
func (c *RPCClient) WANMembers() ([]Member, error) {
	header := requestHeader{
		Command: membersWANCommand,
		Seq:     c.getSeq(),
	}
	var resp membersResponse
	err := c.genericRPC(&header, nil, &resp)
	return resp.Members, err
}
func (c *RPCClient) ListKeys() (keyringResponse, error) {
	header := requestHeader{
		Command: listKeysCommand,
		Seq:     c.getSeq(),
	}
	var resp keyringResponse
	err := c.genericRPC(&header, nil, &resp)
	return resp, err
}
func (c *RPCClient) InstallKey(key string) (keyringResponse, error) {
	header := requestHeader{
		Command: installKeyCommand,
		Seq:     c.getSeq(),
	}
	req := keyringRequest{key}
	var resp keyringResponse
	err := c.genericRPC(&header, &req, &resp)
	return resp, err
}
func (c *RPCClient) UseKey(key string) (keyringResponse, error) {
	header := requestHeader{
		Command: useKeyCommand,
		Seq:     c.getSeq(),
	}
	req := keyringRequest{key}
	var resp keyringResponse
	err := c.genericRPC(&header, &req, &resp)
	return resp, err
}
func (c *RPCClient) RemoveKey(key string) (keyringResponse, error) {
	header := requestHeader{
		Command: removeKeyCommand,
		Seq:     c.getSeq(),
	}
	req := keyringRequest{key}
	var resp keyringResponse
	err := c.genericRPC(&header, &req, &resp)
	return resp, err
}
func (c *RPCClient) Leave() error {
	header := requestHeader{
		Command: leaveCommand,
		Seq:     c.getSeq(),
	}
	return c.genericRPC(&header, nil, nil)
}
func (c *RPCClient) Stats() (map[string]map[string]string, error) {
	header := requestHeader{
		Command: statsCommand,
		Seq:     c.getSeq(),
	}
	var resp map[string]map[string]string
	err := c.genericRPC(&header, nil, &resp)
	return resp, err
}
func (c *RPCClient) Reload() error {
	header := requestHeader{
		Command: reloadCommand,
		Seq:     c.getSeq(),
	}
	return c.genericRPC(&header, nil, nil)
}
type monitorHandler struct {
	client *RPCClient
	closed bool
	init   bool
	initCh chan<- error
	logCh  chan<- string
	seq    uint64
}
func (mh *monitorHandler) Handle(resp *responseHeader) {
	if !mh.init {
		mh.init = true
		mh.initCh <- strToError(resp.Error)
		return
	}
	var rec logRecord
	if err := mh.client.dec.Decode(&rec); err != nil {
		log.Printf("[ERR] Failed to decode log: %v", err)
		mh.client.deregisterHandler(mh.seq)
		return
	}
	select {
	case mh.logCh <- rec.Log:
	default:
		log.Printf("[ERR] Dropping log! Monitor channel full")
	}
}
func (mh *monitorHandler) Cleanup() {
	if !mh.closed {
		if !mh.init {
			mh.init = true
			mh.initCh <- fmt.Errorf("Stream closed")
		}
		close(mh.logCh)
		mh.closed = true
	}
}
func (c *RPCClient) Monitor(level logutils.LogLevel, ch chan<- string) (StreamHandle, error) {
	seq := c.getSeq()
	header := requestHeader{
		Command: monitorCommand,
		Seq:     seq,
	}
	req := monitorRequest{
		LogLevel: string(level),
	}
	initCh := make(chan error, 1)
	handler := &monitorHandler{
		client: c,
		initCh: initCh,
		logCh:  ch,
		seq:    seq,
	}
	c.handleSeq(seq, handler)
	if err := c.send(&header, &req); err != nil {
		c.deregisterHandler(seq)
		return 0, err
	}
	select {
	case err := <-initCh:
		return StreamHandle(seq), err
	case <-c.shutdownCh:
		c.deregisterHandler(seq)
		return 0, clientClosed
	}
}
func (c *RPCClient) Stop(handle StreamHandle) error {
	c.deregisterHandler(uint64(handle))
	header := requestHeader{
		Command: stopCommand,
		Seq:     c.getSeq(),
	}
	req := stopRequest{
		Stop: uint64(handle),
	}
	return c.genericRPC(&header, &req, nil)
}
func (c *RPCClient) handshake() error {
	header := requestHeader{
		Command: handshakeCommand,
		Seq:     c.getSeq(),
	}
	req := handshakeRequest{
		Version: MaxRPCVersion,
	}
	return c.genericRPC(&header, &req, nil)
}
func (c *RPCClient) genericRPC(header *requestHeader, req interface{}, resp interface{}) error {
	errCh := make(chan error, 1)
	handler := func(respHeader *responseHeader) {
		if resp != nil {
			err := c.dec.Decode(resp)
			if err != nil {
				errCh <- err
				return
			}
		}
		errCh <- strToError(respHeader.Error)
	}
	c.handleSeq(header.Seq, &seqCallback{handler: handler})
	defer c.deregisterHandler(header.Seq)
	if err := c.send(header, req); err != nil {
		return err
	}
	select {
	case err := <-errCh:
		return err
	case <-c.shutdownCh:
		return clientClosed
	}
}
func strToError(s string) error {
	if s != "" {
		return fmt.Errorf(s)
	}
	return nil
}
func (c *RPCClient) getSeq() uint64 {
	return atomic.AddUint64(&c.seq, 1)
}
func (c *RPCClient) deregisterAll() {
	c.dispatchLock.Lock()
	defer c.dispatchLock.Unlock()
	for _, seqH := range c.dispatch {
		seqH.Cleanup()
	}
	c.dispatch = make(map[uint64]seqHandler)
}
func (c *RPCClient) deregisterHandler(seq uint64) {
	c.dispatchLock.Lock()
	seqH, ok := c.dispatch[seq]
	delete(c.dispatch, seq)
	c.dispatchLock.Unlock()
	if ok {
		seqH.Cleanup()
	}
}