func (n *netPipeline) Consumer() <-chan AppendFuture {
	return n.doneCh
}
Consumer returns a channel that can be used to consume complete futures
package raft
import (
	"bufio"
	"errors"
	"fmt"
	"io"
	"log"
	"net"
	"os"
	"sync"
	"time"
	"github.com/socketplane/socketplane/Godeps/_workspace/src/github.com/hashicorp/go-msgpack/codec"
)
const (
	rpcAppendEntries uint8 = iota
	rpcRequestVote
	rpcInstallSnapshot
	DefaultTimeoutScale = 256 * 1024  
	rpcMaxPipeline = 128
)
var (
	ErrTransportShutdown = errors.New("transport shutdown")
	ErrPipelineShutdown = errors.New("append pipeline closed")
)
type NetworkTransport struct {
	connPool     map[string][]*netConn
	connPoolLock sync.Mutex
	consumeCh chan RPC
	heartbeatFn     func(RPC)
	heartbeatFnLock sync.Mutex
	logger *log.Logger
	maxPool int
	shutdown     bool
	shutdownCh   chan struct{}
	shutdownLock sync.Mutex
	stream StreamLayer
	timeout      time.Duration
	TimeoutScale int
}
type StreamLayer interface {
	net.Listener
	Dial(address string, timeout time.Duration) (net.Conn, error)
}
type netConn struct {
	target net.Addr
	conn   net.Conn
	r      *bufio.Reader
	w      *bufio.Writer
	dec    *codec.Decoder
	enc    *codec.Encoder
}
func (n *netConn) Release() error {
	return n.conn.Close()
}
type netPipeline struct {
	conn  *netConn
	trans *NetworkTransport
	doneCh       chan AppendFuture
	inprogressCh chan *appendFuture
	shutdown     bool
	shutdownCh   chan struct{}
	shutdownLock sync.Mutex
}
func NewNetworkTransport(
	stream StreamLayer,
	maxPool int,
	timeout time.Duration,
	logOutput io.Writer,
) *NetworkTransport {
	if logOutput == nil {
		logOutput = os.Stderr
	}
	trans := &NetworkTransport{
		connPool:     make(map[string][]*netConn),
		consumeCh:    make(chan RPC),
		logger:       log.New(logOutput, "", log.LstdFlags),
		maxPool:      maxPool,
		shutdownCh:   make(chan struct{}),
		stream:       stream,
		timeout:      timeout,
		TimeoutScale: DefaultTimeoutScale,
	}
	go trans.listen()
	return trans
}
func (n *NetworkTransport) SetHeartbeatHandler(cb func(rpc RPC)) {
	n.heartbeatFnLock.Lock()
	defer n.heartbeatFnLock.Unlock()
	n.heartbeatFn = cb
}
func (n *NetworkTransport) Close() error {
	n.shutdownLock.Lock()
	defer n.shutdownLock.Unlock()
	if !n.shutdown {
		close(n.shutdownCh)
		n.stream.Close()
		n.shutdown = true
	}
	return nil
}
func (n *NetworkTransport) Consumer() <-chan RPC {
	return n.consumeCh
}
func (n *NetworkTransport) LocalAddr() net.Addr {
	return n.stream.Addr()
}
func (n *NetworkTransport) IsShutdown() bool {
	select {
	case <-n.shutdownCh:
		return true
	default:
		return false
	}
}
func (n *NetworkTransport) getPooledConn(target net.Addr) *netConn {
	n.connPoolLock.Lock()
	defer n.connPoolLock.Unlock()
	key := target.String()
	conns, ok := n.connPool[key]
	if !ok || len(conns) == 0 {
		return nil
	}
	var conn *netConn
	num := len(conns)
	conn, conns[num-1] = conns[num-1], nil
	n.connPool[key] = conns[:num-1]
	return conn
}
func (n *NetworkTransport) getConn(target net.Addr) (*netConn, error) {
	if conn := n.getPooledConn(target); conn != nil {
		return conn, nil
	}
	conn, err := n.stream.Dial(target.String(), n.timeout)
	if err != nil {
		return nil, err
	}
	netConn := &netConn{
		target: target,
		conn:   conn,
		r:      bufio.NewReader(conn),
		w:      bufio.NewWriter(conn),
	}
	netConn.dec = codec.NewDecoder(netConn.r, &codec.MsgpackHandle{})
	netConn.enc = codec.NewEncoder(netConn.w, &codec.MsgpackHandle{})
	return netConn, nil
}
func (n *NetworkTransport) returnConn(conn *netConn) {
	n.connPoolLock.Lock()
	defer n.connPoolLock.Unlock()
	key := conn.target.String()
	conns, _ := n.connPool[key]
	if !n.IsShutdown() && len(conns) < n.maxPool {
		n.connPool[key] = append(conns, conn)
	} else {
		conn.Release()
	}
}
func (n *NetworkTransport) AppendEntriesPipeline(target net.Addr) (AppendPipeline, error) {
	conn, err := n.getConn(target)
	if err != nil {
		return nil, err
	}
	return newNetPipeline(n, conn), nil
}
func (n *NetworkTransport) AppendEntries(target net.Addr, args *AppendEntriesRequest, resp *AppendEntriesResponse) error {
	return n.genericRPC(target, rpcAppendEntries, args, resp)
}
func (n *NetworkTransport) RequestVote(target net.Addr, args *RequestVoteRequest, resp *RequestVoteResponse) error {
	return n.genericRPC(target, rpcRequestVote, args, resp)
}
func (n *NetworkTransport) genericRPC(target net.Addr, rpcType uint8, args interface{}, resp interface{}) error {
	conn, err := n.getConn(target)
	if err != nil {
		return err
	}
	if n.timeout > 0 {
		conn.conn.SetDeadline(time.Now().Add(n.timeout))
	}
	if err := sendRPC(conn, rpcType, args); err != nil {
		return err
	}
	canReturn, err := decodeResponse(conn, resp)
	if canReturn {
		n.returnConn(conn)
	}
	return err
}
func (n *NetworkTransport) InstallSnapshot(target net.Addr, args *InstallSnapshotRequest, resp *InstallSnapshotResponse, data io.Reader) error {
	conn, err := n.getConn(target)
	if err != nil {
		return err
	}
	defer conn.Release()
	if n.timeout > 0 {
		timeout := n.timeout * time.Duration(args.Size/int64(n.TimeoutScale))
		if timeout < n.timeout {
			timeout = n.timeout
		}
		conn.conn.SetDeadline(time.Now().Add(timeout))
	}
	if err := sendRPC(conn, rpcInstallSnapshot, args); err != nil {
		return err
	}
	if _, err := io.Copy(conn.w, data); err != nil {
		return err
	}
	if err := conn.w.Flush(); err != nil {
		return err
	}
	_, err = decodeResponse(conn, resp)
	return err
}
func (n *NetworkTransport) EncodePeer(p net.Addr) []byte {
	return []byte(p.String())
}
func (n *NetworkTransport) DecodePeer(buf []byte) net.Addr {
	addr, err := net.ResolveTCPAddr("tcp", string(buf))
	if err != nil {
		panic(fmt.Errorf("failed to parse network address: %s", buf))
	}
	return addr
}
func (n *NetworkTransport) listen() {
	for {
		conn, err := n.stream.Accept()
		if err != nil {
			if n.IsShutdown() {
				return
			}
			n.logger.Printf("[ERR] raft-net: Failed to accept connection: %v", err)
			continue
		}
		n.logger.Printf("[DEBUG] raft-net: %v accepted connection from: %v", n.LocalAddr(), conn.RemoteAddr())
		go n.handleConn(conn)
	}
}
func (n *NetworkTransport) handleConn(conn net.Conn) {
	defer conn.Close()
	r := bufio.NewReader(conn)
	w := bufio.NewWriter(conn)
	dec := codec.NewDecoder(r, &codec.MsgpackHandle{})
	enc := codec.NewEncoder(w, &codec.MsgpackHandle{})
	for {
		if err := n.handleCommand(r, dec, enc); err != nil {
			if err != io.EOF {
				n.logger.Printf("[ERR] raft-net: Failed to decode incoming command: %v", err)
			}
			return
		}
		if err := w.Flush(); err != nil {
			n.logger.Printf("[ERR] raft-net: Failed to flush response: %v", err)
			return
		}
	}
}
func (n *NetworkTransport) handleCommand(r *bufio.Reader, dec *codec.Decoder, enc *codec.Encoder) error {
	rpcType, err := r.ReadByte()
	if err != nil {
		return err
	}
	respCh := make(chan RPCResponse, 1)
	rpc := RPC{
		RespChan: respCh,
	}
	isHeartbeat := false
	switch rpcType {
	case rpcAppendEntries:
		var req AppendEntriesRequest
		if err := dec.Decode(&req); err != nil {
			return err
		}
		rpc.Command = &req
		if req.Term != 0 && req.Leader != nil &&
			req.PrevLogEntry == 0 && req.PrevLogTerm == 0 &&
			len(req.Entries) == 0 && req.LeaderCommitIndex == 0 {
			isHeartbeat = true
		}
	case rpcRequestVote:
		var req RequestVoteRequest
		if err := dec.Decode(&req); err != nil {
			return err
		}
		rpc.Command = &req
	case rpcInstallSnapshot:
		var req InstallSnapshotRequest
		if err := dec.Decode(&req); err != nil {
			return err
		}
		rpc.Command = &req
		rpc.Reader = io.LimitReader(r, req.Size)
	default:
		return fmt.Errorf("unknown rpc type %d", rpcType)
	}
	if isHeartbeat {
		n.heartbeatFnLock.Lock()
		fn := n.heartbeatFn
		n.heartbeatFnLock.Unlock()
		if fn != nil {
			fn(rpc)
			goto RESP
		}
	}
	select {
	case n.consumeCh <- rpc:
	case <-n.shutdownCh:
		return ErrTransportShutdown
	}
RESP:
	select {
	case resp := <-respCh:
		respErr := ""
		if resp.Error != nil {
			respErr = resp.Error.Error()
		}
		if err := enc.Encode(respErr); err != nil {
			return err
		}
		if err := enc.Encode(resp.Response); err != nil {
			return err
		}
	case <-n.shutdownCh:
		return ErrTransportShutdown
	}
	return nil
}
func decodeResponse(conn *netConn, resp interface{}) (bool, error) {
	var rpcError string
	if err := conn.dec.Decode(&rpcError); err != nil {
		conn.Release()
		return false, err
	}
	if err := conn.dec.Decode(resp); err != nil {
		conn.Release()
		return false, err
	}
	if rpcError != "" {
		return true, fmt.Errorf(rpcError)
	}
	return true, nil
}
func sendRPC(conn *netConn, rpcType uint8, args interface{}) error {
	if err := conn.w.WriteByte(rpcType); err != nil {
		conn.Release()
		return err
	}
	if err := conn.enc.Encode(args); err != nil {
		conn.Release()
		return err
	}
	if err := conn.w.Flush(); err != nil {
		conn.Release()
		return err
	}
	return nil
}
func newNetPipeline(trans *NetworkTransport, conn *netConn) *netPipeline {
	n := &netPipeline{
		conn:         conn,
		trans:        trans,
		doneCh:       make(chan AppendFuture, rpcMaxPipeline),
		inprogressCh: make(chan *appendFuture, rpcMaxPipeline),
		shutdownCh:   make(chan struct{}),
	}
	go n.decodeResponses()
	return n
}
func (n *netPipeline) decodeResponses() {
	timeout := n.trans.timeout
	for {
		select {
		case future := <-n.inprogressCh:
			if timeout > 0 {
				n.conn.conn.SetReadDeadline(time.Now().Add(timeout))
			}
			_, err := decodeResponse(n.conn, future.resp)
			future.respond(err)
			select {
			case n.doneCh <- future:
			case <-n.shutdownCh:
				return
			}
		case <-n.shutdownCh:
			return
		}
	}
}
func (n *netPipeline) AppendEntries(args *AppendEntriesRequest, resp *AppendEntriesResponse) (AppendFuture, error) {
	future := &appendFuture{
		start: time.Now(),
		args:  args,
		resp:  resp,
	}
	future.init()
	if timeout := n.trans.timeout; timeout > 0 {
		n.conn.conn.SetWriteDeadline(time.Now().Add(timeout))
	}
	if err := sendRPC(n.conn, rpcAppendEntries, future.args); err != nil {
		return nil, err
	}
	select {
	case n.inprogressCh <- future:
		return future, nil
	case <-n.shutdownCh:
		return nil, ErrPipelineShutdown
	}
}