luxolus
/
open-consul
2
0
Fork 0
Code Issues 1 Pull Requests Packages Releases Wiki Activity

Merge pull request #199 from abursavich/master 

reuse codec.MsgpackHandle
This commit is contained in:
Armon Dadgar 2014-06-08 13:52:14 -07:00
parent 7fc5b7a920 d209517d50
commit 3e2dd39baf
8 changed files with 19 additions and 25 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
command/agent/rpc.go
View File

@ -60,6 +60,8 @@ const (
monitorExists = "Monitor already exists"
)
var mh = codec.MsgpackHandle{RawToString: true, WriteExt: true}
// Request header is sent before each request
type requestHeader struct {
Command string
@ -249,10 +251,8 @@ func (i *AgentRPC) listen() {
reader: bufio.NewReader(conn),
writer: bufio.NewWriter(conn),
}
client.dec = codec.NewDecoder(client.reader,
&codec.MsgpackHandle{RawToString: true, WriteExt: true})
client.enc = codec.NewEncoder(client.writer,
&codec.MsgpackHandle{RawToString: true, WriteExt: true})
client.dec = codec.NewDecoder(client.reader, &mh)
client.enc = codec.NewEncoder(client.writer, &mh)
if err != nil {
i.logger.Printf("[ERR] agent.rpc: Failed to create decoder: %v", err)
conn.Close()

6
command/agent/rpc_client.go
View File

@ -94,10 +94,8 @@ func NewRPCClient(addr string) (*RPCClient, error) {
dispatch: make(map[uint64]seqHandler),
shutdownCh: make(chan struct{}),
}
client.dec = codec.NewDecoder(client.reader,
&codec.MsgpackHandle{RawToString: true, WriteExt: true})
client.enc = codec.NewEncoder(client.writer,
&codec.MsgpackHandle{RawToString: true, WriteExt: true})
client.dec = codec.NewDecoder(client.reader, &mh)
client.enc = codec.NewEncoder(client.writer, &mh)
go client.listen()
// Do the initial handshake

6
consul/fsm.go
View File

@ -221,8 +221,7 @@ func (c *consulFSM) Restore(old io.ReadCloser) error {
c.state = state
// Create a decoder
var handle codec.MsgpackHandle
dec := codec.NewDecoder(old, &handle)
dec := codec.NewDecoder(old, &mh)
// Read in the header
var header snapshotHeader
@ -278,8 +277,7 @@ func (c *consulFSM) Restore(old io.ReadCloser) error {
func (s *consulSnapshot) Persist(sink raft.SnapshotSink) error {
// Register the nodes
handle := codec.MsgpackHandle{}
encoder := codec.NewEncoder(sink, &handle)
encoder := codec.NewEncoder(sink, &mh)
// Write the header
header := snapshotHeader{

6
consul/mdb_table_test.go
View File

@ -19,8 +19,7 @@ type MockData struct {
func MockEncoder(obj interface{}) []byte {
buf := bytes.NewBuffer(nil)
handle := codec.MsgpackHandle{}
encoder := codec.NewEncoder(buf, &handle)
encoder := codec.NewEncoder(buf, &mh)
err := encoder.Encode(obj)
if err != nil {
panic(err)
@ -30,8 +29,7 @@ func MockEncoder(obj interface{}) []byte {
func MockDecoder(buf []byte) interface{} {
out := new(MockData)
var handle codec.MsgpackHandle
err := codec.NewDecoder(bytes.NewReader(buf), &handle).Decode(out)
err := codec.NewDecoder(bytes.NewReader(buf), &mh).Decode(out)
if err != nil {
panic(err)
}

4
consul/pool.go
View File

@ -15,6 +15,8 @@ import (
"time"
)
var mh = codec.MsgpackHandle{}
// muxSession is used to provide an interface for either muxado or yamux
type muxSession interface {
Open() (net.Conn, error)
@ -79,7 +81,7 @@ func (c *Conn) getClient() (*StreamClient, error) {
}
// Create the RPC client
cc := codec.GoRpc.ClientCodec(stream, &codec.MsgpackHandle{})
cc := codec.GoRpc.ClientCodec(stream, &mh)
client := rpc.NewClientWithCodec(cc)
// Return a new stream client

2
consul/rpc.go
View File

@ -148,7 +148,7 @@ func (s *Server) handleMultiplexV2(conn net.Conn) {
// handleConsulConn is used to service a single Consul RPC connection
func (s *Server) handleConsulConn(conn net.Conn) {
defer conn.Close()
rpcCodec := codec.GoRpc.ServerCodec(conn, &codec.MsgpackHandle{})
rpcCodec := codec.GoRpc.ServerCodec(conn, &mh)
for !s.shutdown {
if err := s.rpcServer.ServeRequest(rpcCodec); err != nil {
if err != io.EOF && !strings.Contains(err.Error(), "closed") {

2
consul/status_endpoint_test.go
View File

@ -19,7 +19,7 @@ func rpcClient(t *testing.T, s *Server) *rpc.Client {
// Write the Consul RPC byte to set the mode
conn.Write([]byte{byte(rpcConsul)})
cc := codec.GoRpc.ClientCodec(conn, &codec.MsgpackHandle{})
cc := codec.GoRpc.ClientCodec(conn, &mh)
return rpc.NewClientWithCodec(cc)
}

10
consul/structs/structs.go
View File

@ -395,19 +395,17 @@ type IndexedSessions struct {
QueryMeta
}
var mh = codec.MsgpackHandle{}
// Decode is used to decode a MsgPack encoded object
func Decode(buf []byte, out interface{}) error {
var handle codec.MsgpackHandle
return codec.NewDecoder(bytes.NewReader(buf), &handle).Decode(out)
return codec.NewDecoder(bytes.NewReader(buf), &mh).Decode(out)
}
// Encode is used to encode a MsgPack object with type prefix
func Encode(t MessageType, msg interface{}) ([]byte, error) {
buf := bytes.NewBuffer(nil)
buf.WriteByte(uint8(t))
handle := codec.MsgpackHandle{}
encoder := codec.NewEncoder(buf, &handle)
err := encoder.Encode(msg)
err := codec.NewEncoder(buf, &mh).Encode(msg)
return buf.Bytes(), err
}