func (c *NamenodeConnection) readResponse(method string, resp proto.Message) error {
	var packetLength uint32
	err := binary.Read(c.conn, binary.BigEndian, &packetLength)
	if err != nil {
		return err
	}
	packet := make([]byte, packetLength)
	_, err = io.ReadFull(c.conn, packet)
	if err != nil {
		return err
	}
	rrh := &hadoop.RpcResponseHeaderProto{}
	err = readRPCPacket(packet, rrh, resp)
	if rrh.GetStatus() != hadoop.RpcResponseHeaderProto_SUCCESS {
		return &NamenodeError{
			Method:    method,
			Message:   rrh.GetErrorMsg(),
			Code:      int(rrh.GetErrorDetail()),
			Exception: rrh.GetExceptionClassName(),
		}
	} else if int(rrh.GetCallId()) != c.currentRequestID {
		return errors.New("Error reading response: unexpected sequence number")
	}
	return nil
}
A response from the namenode:
+-----------------------------------------------------------+
|  uint32 length of the next two parts                      |
+-----------------------------------------------------------+
|  varint length + RpcResponseHeaderProto                   |
+-----------------------------------------------------------+
|  varint length + Response                                 |
+-----------------------------------------------------------+
package rpc
import (
	"encoding/binary"
	"errors"
	"fmt"
	"io"
	"net"
	"sync"
	hadoop "github.com/colinmarc/hdfs/protocol/hadoop_common"
	"github.com/golang/protobuf/proto"
)
const (
	rpcVersion           = 0x09
	serviceClass         = 0x0
	authProtocol         = 0x0
	protocolClass        = "org.apache.hadoop.hdfs.protocol.ClientProtocol"
	protocolClassVersion = 1
	handshakeCallID      = -3
)
type NamenodeConnection struct {
	clientId         []byte
	clientName       string
	currentRequestID int
	user             string
	conn             net.Conn
	reqLock          sync.Mutex
}
type NamenodeError struct {
	Method    string
	Message   string
	Code      int
	Exception string
}
func (err *NamenodeError) Desc() string {
	return hadoop.RpcResponseHeaderProto_RpcErrorCodeProto_name[int32(err.Code)]
}
func (err *NamenodeError) Error() string {
	s := fmt.Sprintf("%s call failed with %s", err.Method, err.Desc())
	if err.Exception != "" {
		s += fmt.Sprintf(" (%s)", err.Exception)
	}
	return s
}
func NewNamenodeConnection(address, user string) (*NamenodeConnection, error) {
	conn, err := net.DialTimeout("tcp", address, connectTimeout)
	if err != nil {
		return nil, err
	}
	return WrapNamenodeConnection(conn, user)
}
func WrapNamenodeConnection(conn net.Conn, user string) (*NamenodeConnection, error) {
	clientId := newClientID()
	c := &NamenodeConnection{
		clientId:   clientId,
		clientName: "go-hdfs-" + string(clientId),
		user:       user,
		conn:       conn,
	}
	err := c.writeNamenodeHandshake()
	if err != nil {
		conn.Close()
		return nil, fmt.Errorf("Error performing handshake: %s", err)
	}
	return c, nil
}
func (c *NamenodeConnection) ClientName() string {
	return c.clientName
}
func (c *NamenodeConnection) Execute(method string, req proto.Message, resp proto.Message) error {
	c.reqLock.Lock()
	defer c.reqLock.Unlock()
	c.currentRequestID++
	err := c.writeRequest(method, req)
	if err != nil {
		c.conn.Close()
		return err
	}
	err = c.readResponse(method, resp)
	if err != nil {
		if _, ok := err.(*NamenodeError); !ok {
			c.conn.Close()  
		}
		return err
	}
	return nil
}
func (c *NamenodeConnection) writeRequest(method string, req proto.Message) error {
	rrh := newRPCRequestHeader(c.currentRequestID, c.clientId)
	rh := newRequestHeader(method)
	reqBytes, err := makeRPCPacket(rrh, rh, req)
	if err != nil {
		return err
	}
	_, err = c.conn.Write(reqBytes)
	return err
}