func (n *NetworkTransport) LocalAddr() net.Addr {
	return n.stream.Addr()
}
LocalAddr implements the Transport interface.
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
}