func writerPipe(conn *net.TCPConn) {
	for {
		var msg string
		fmt.Scanln(&msg)

		if strings.EqualFold(msg, "quit") {
			quitSp <- true
			break
		}
		testMessage := &pb.TestMessage{
			TestInt:    proto.Int32(123),
			TestString: proto.String("est"),
		}
		fmt.Println(testMessage)
		byt, _ := proto.Marshal(testMessage)
		fmt.Println(byt)

		buff := &pb.MobileSuiteProtobuf{
			Type:    proto.Int32(321),
			Arena:   proto.Int32(111),
			Command: proto.Int32(0xa),
			Message: byt,
		}
		fmt.Println(buff)
		bybuf, _ := proto.Marshal(buff)
		fmt.Println(bybuf)
		conn.Write(bybuf)

	}
}

// Insert can insert record into a btree
func (t *Btree) insert(record TreeLog) error {
	tnode, err := t.getTreeNode(t.GetRoot())
	if err != nil {
		if err.Error() != "no data" {
			return err
		}
		nnode := t.newTreeNode()
		nnode.NodeType = proto.Int32(isLeaf)
		_, err = nnode.insertRecord(record, t)
		if err == nil {
			t.Nodes[nnode.GetId()], err = proto.Marshal(nnode)
		}
		t.Root = proto.Int64(nnode.GetId())
		return err
	}
	clonednode, err := tnode.insertRecord(record, t)
	if err == nil && len(clonednode.GetKeys()) > int(t.GetNodeMax()) {
		nnode := t.newTreeNode()
		nnode.NodeType = proto.Int32(isNode)
		key, left, right := clonednode.split(t)
		nnode.insertOnce(key, left, right, t)
		t.Nodes[nnode.GetId()], err = proto.Marshal(nnode)
		t.Root = proto.Int64(nnode.GetId())
	} else {
		t.Root = proto.Int64(clonednode.GetId())
	}
	return err
}

// WriteRequest - implement rpc.ClientCodec interface.
func (c *pbClientCodec) WriteRequest(r *rpc.Request, body interface{}) (err error) {
	// Use a mutex to guarantee the header/body are written in the correct order.
	c.mu.Lock()
	defer c.mu.Unlock()

	// This is protobuf, of course we copy it.
	pbr := &Request{ServiceMethod: &r.ServiceMethod, Seq: &r.Seq}
	data, err := proto.Marshal(pbr)
	if err != nil {
		return
	}
	_, err = WriteNetString(c.rwc, data)
	if err != nil {
		return
	}

	// Of course this is a protobuf! Trust me or detonate the program.
	data, err = proto.Marshal(body.(proto.Message))
	if err != nil {
		return
	}
	_, err = WriteNetString(c.rwc, data)
	if err != nil {
		return
	}

	if flusher, ok := c.rwc.(flusher); ok {
		err = flusher.Flush()
	}
	return
}

func (c *clientCodec) WriteRequest(rpcreq *rpc.Request, param interface{}) error {
	rr := *rpcreq
	req := &Request{}

	c.mutex.Lock()
	req.Id = proto.Uint64(c.next)
	c.next++
	c.pending[*req.Id] = &rr
	c.mutex.Unlock()

	req.Method = proto.String(rpcreq.ServiceMethod)
	if msg, ok := param.(proto.Message); ok {
		body, err := proto.Marshal(msg)
		if err != nil {
			return err
		}
		req.Body = body
	} else {
		return fmt.Errorf("marshal request param error: %s", param)
	}

	f, err := proto.Marshal(req)
	if err != nil {
		return err
	}

	if err := write(c.w, f); err != nil {
		return err
	}

	return nil
}

// WriteResponse writes a response on the codec.
func (c *pbServerCodec) WriteResponse(r *rpc.Response, body interface{}, last bool) (err error) {
	// Use a mutex to guarantee the header/body are written in the correct order.
	c.mu.Lock()
	defer c.mu.Unlock()
	rtmp := &Response{ServiceMethod: &r.ServiceMethod, Seq: &r.Seq, Error: &r.Error}
	data, err := proto.Marshal(rtmp)
	if err != nil {
		return
	}
	_, err = WriteNetString(c.rwc, data)
	if err != nil {
		return
	}

	if pb, ok := body.(proto.Message); ok {
		data, err = proto.Marshal(pb)
		if err != nil {
			return
		}
	} else {
		data = nil
	}
	_, err = WriteNetString(c.rwc, data)
	if err != nil {
		return
	}

	if flusher, ok := c.rwc.(flusher); ok {
		err = flusher.Flush()
	}
	return
}

//处理查找节点的请求
//本节点定期查询已知节点是否在线，更新节点信息
func (this *Manager) read() {
	clientConn, _ := this.serverManager.GetController().GetSession(this.rootId.String())
	for {
		node := <-this.nodeStore.OutFindNode
		if node.NodeId != nil {
			findNodeOne := &msg.FindNodeReq{
				NodeId: proto.String(this.nodeStore.GetRootId()),
				FindId: proto.String(node.NodeId.String()),
			}
			findNodeBytes, _ := proto.Marshal(findNodeOne)
			// clientConn := this.serverManager.GetController().GetClientByName("firstConnPeer")
			// fmt.Println(clientConn)
			clientConn.Send(msg.FindNodeReqNum, &findNodeBytes)
		}
		if node.NodeIdShould != nil {
			findNodeOne := &msg.FindNodeReq{
				NodeId: proto.String(this.nodeStore.GetRootId()),
				FindId: proto.String(node.NodeIdShould.String()),
			}
			findNodeBytes, _ := proto.Marshal(findNodeOne)
			// clientConn := this.serverManager.GetController().GetClientByName("firstConnPeer")
			clientConn.Send(msg.FindNodeReqNum, &findNodeBytes)
		}
	}
}

func sendRequest(c *Client, conn *connection, rpcCall *call) error {
	//log.Println("About to call RPC: ", rpcCall.procedure)

	// 0. RpcRequestHeaderProto
	var clientId [16]byte = [16]byte(*c.ClientId)
	rpcReqHeaderProto := hadoop_common.RpcRequestHeaderProto{RpcKind: &gohadoop.RPC_PROTOCOL_BUFFFER, RpcOp: &gohadoop.RPC_FINAL_PACKET, CallId: &rpcCall.callId, ClientId: clientId[0:16], RetryCount: &rpcCall.retryCount}
	rpcReqHeaderProtoBytes, err := proto.Marshal(&rpcReqHeaderProto)
	if err != nil {
		log.Fatal("proto.Marshal(&rpcReqHeaderProto)", err)
		return err
	}

	// 1. RequestHeaderProto
	requestHeaderProto := rpcCall.procedure
	requestHeaderProtoBytes, err := proto.Marshal(requestHeaderProto)
	if err != nil {
		log.Fatal("proto.Marshal(&requestHeaderProto)", err)
		return err
	}

	// 2. Param
	paramProto := rpcCall.request
	paramProtoBytes, err := proto.Marshal(paramProto)
	if err != nil {
		log.Fatal("proto.Marshal(&paramProto)", err)
		return err
	}

	totalLength := len(rpcReqHeaderProtoBytes) + sizeVarint(len(rpcReqHeaderProtoBytes)) + len(requestHeaderProtoBytes) + sizeVarint(len(requestHeaderProtoBytes)) + len(paramProtoBytes) + sizeVarint(len(paramProtoBytes))
	var tLen int32 = int32(totalLength)
	if totalLengthBytes, err := gohadoop.ConvertFixedToBytes(tLen); err != nil {
		log.Fatal("ConvertFixedToBytes(totalLength)", err)
		return err
	} else {
		if _, err := conn.con.Write(totalLengthBytes); err != nil {
			log.Fatal("conn.con.Write(totalLengthBytes)", err)
			return err
		}
	}

	if err := writeDelimitedBytes(conn, rpcReqHeaderProtoBytes); err != nil {
		log.Fatal("writeDelimitedBytes(conn, rpcReqHeaderProtoBytes)", err)
		return err
	}
	if err := writeDelimitedBytes(conn, requestHeaderProtoBytes); err != nil {
		log.Fatal("writeDelimitedBytes(conn, requestHeaderProtoBytes)", err)
		return err
	}
	if err := writeDelimitedBytes(conn, paramProtoBytes); err != nil {
		log.Fatal("writeDelimitedBytes(conn, paramProtoBytes)", err)
		return err
	}

	//log.Println("Succesfully sent request of length: ", totalLength)

	return nil
}

func (c *context) Call(service, method string, in, out proto.Message, opts *internal.CallOptions) error {
	req, err := proto.Marshal(in)
	if err != nil {
		return fmt.Errorf("error marshalling request: %v", err)
	}

	remReq := &pb.Request{
		ServiceName: proto.String(service),
		Method:      proto.String(method),
		Request:     req,
		// NOTE(djd): RequestId is unused in the server.
	}

	req, err = proto.Marshal(remReq)
	if err != nil {
		return fmt.Errorf("proto.Marshal: %v", err)
	}

	// TODO(djd): Respect opts.Timeout?
	resp, err := c.client.Post(c.url, "application/octet-stream", bytes.NewReader(req))
	if err != nil {
		return fmt.Errorf("error sending request: %v", err)
	}
	defer resp.Body.Close()

	body, err := ioutil.ReadAll(resp.Body)
	if resp.StatusCode != http.StatusOK {
		return fmt.Errorf("bad response %d; body: %q", resp.StatusCode, body)
	}
	if err != nil {
		return fmt.Errorf("failed reading response: %v", err)
	}
	remResp := &pb.Response{}
	if err := proto.Unmarshal(body, remResp); err != nil {
		return fmt.Errorf("error unmarshalling response: %v", err)
	}

	if ae := remResp.GetApplicationError(); ae != nil {
		return &internal.APIError{
			Code:    ae.GetCode(),
			Detail:  ae.GetDetail(),
			Service: service,
		}
	}

	if remResp.Response == nil {
		return fmt.Errorf("unexpected response: %s", proto.MarshalTextString(remResp))
	}

	return proto.Unmarshal(remResp.Response, out)
}

func writeConnectionContext(c *Client, conn *connection, connectionId *connection_id, authProtocol gohadoop.AuthProtocol) error {
	// Create hadoop_common.IpcConnectionContextProto
	ugi, _ := gohadoop.CreateSimpleUGIProto()
	ipcCtxProto := hadoop_common.IpcConnectionContextProto{UserInfo: ugi, Protocol: &connectionId.protocol}

	// Create RpcRequestHeaderProto
	var callId int32 = -3
	var clientId [16]byte = [16]byte(*c.ClientId)

	/*if (authProtocol == gohadoop.AUTH_PROTOCOL_SASL) {
	  callId = SASL_RPC_CALL_ID
	}*/

	rpcReqHeaderProto := hadoop_common.RpcRequestHeaderProto{RpcKind: &gohadoop.RPC_PROTOCOL_BUFFFER, RpcOp: &gohadoop.RPC_FINAL_PACKET, CallId: &callId, ClientId: clientId[0:16], RetryCount: &gohadoop.RPC_DEFAULT_RETRY_COUNT}

	rpcReqHeaderProtoBytes, err := proto.Marshal(&rpcReqHeaderProto)
	if err != nil {
		log.Fatal("proto.Marshal(&rpcReqHeaderProto)", err)
		return err
	}

	ipcCtxProtoBytes, _ := proto.Marshal(&ipcCtxProto)
	if err != nil {
		log.Fatal("proto.Marshal(&ipcCtxProto)", err)
		return err
	}

	totalLength := len(rpcReqHeaderProtoBytes) + sizeVarint(len(rpcReqHeaderProtoBytes)) + len(ipcCtxProtoBytes) + sizeVarint(len(ipcCtxProtoBytes))
	var tLen int32 = int32(totalLength)
	totalLengthBytes, err := gohadoop.ConvertFixedToBytes(tLen)

	if err != nil {
		log.Fatal("ConvertFixedToBytes(totalLength)", err)
		return err
	} else if _, err := conn.con.Write(totalLengthBytes); err != nil {
		log.Fatal("conn.con.Write(totalLengthBytes)", err)
		return err
	}

	if err := writeDelimitedBytes(conn, rpcReqHeaderProtoBytes); err != nil {
		log.Fatal("writeDelimitedBytes(conn, rpcReqHeaderProtoBytes)", err)
		return err
	}
	if err := writeDelimitedBytes(conn, ipcCtxProtoBytes); err != nil {
		log.Fatal("writeDelimitedBytes(conn, ipcCtxProtoBytes)", err)
		return err
	}

	return nil
}

func runAvailabilityPipelineAugmented(startTimestamp int64, timestamps map[string]int64, moreTimestamps map[string]int64) {
	levelDbManager := store.NewSliceManager()

	tracesStore := levelDbManager.Writer("traces")
	tracesStore.BeginWriting()
	for encodedKey, timestamp := range timestamps {
		trace := Trace{
			TraceCreationTimestamp: proto.Int64(timestamp),
		}
		encodedTrace, err := proto.Marshal(&trace)
		if err != nil {
			panic(fmt.Errorf("Error encoding protocol buffer: %v", err))
		}
		tracesStore.WriteRecord(&store.Record{Key: []byte(encodedKey), Value: encodedTrace})
	}
	tracesStore.EndWriting()

	writer := bytes.NewBuffer([]byte{})
	transformer.RunPipeline(AvailabilityPipeline(levelDbManager, writer, startTimestamp))

	tracesStore.BeginWriting()
	for encodedKey, timestamp := range moreTimestamps {
		trace := Trace{
			TraceCreationTimestamp: proto.Int64(timestamp),
		}
		encodedTrace, err := proto.Marshal(&trace)
		if err != nil {
			panic(fmt.Errorf("Error encoding protocol buffer: %v", err))
		}
		tracesStore.WriteRecord(&store.Record{Key: []byte(encodedKey), Value: encodedTrace})
	}
	tracesStore.EndWriting()

	anotherTracesSlice := make([]*store.Record, 0)
	for encodedKey, timestamp := range moreTimestamps {
		trace := Trace{
			TraceCreationTimestamp: proto.Int64(timestamp),
		}
		encodedTrace, err := proto.Marshal(&trace)
		if err != nil {
			panic(fmt.Errorf("Error encoding protocol buffer: %v", err))
		}
		anotherTracesSlice = append(anotherTracesSlice, &store.Record{Key: []byte(encodedKey), Value: encodedTrace})
	}

	anotherWriter := bytes.NewBuffer([]byte{})
	transformer.RunPipeline(AvailabilityPipeline(levelDbManager, anotherWriter, startTimestamp))
	fmt.Printf("%s", anotherWriter.Bytes())
}

func (n *TreeNode) merge(tree *Btree, index int) int64 {
	left, err := tree.getTreeNode(n.Childrens[index])
	if err != nil {
		return -1
	}
	right, err := tree.getTreeNode(n.Childrens[index+1])
	if err != nil {
		return -1
	}
	if len(left.Keys)+len(right.Keys) > int(tree.GetNodeMax()) {
		return -1
	}
	if (len(left.Values) + len(right.Values)) > int(tree.GetLeafMax()) {
		return -1
	}
	leftClone := left.clone(tree)
	n.Childrens[index] = leftClone.GetId()
	if leftClone.GetNodeType() == isLeaf {
		if index == len(n.Keys) {
			n.Childrens = n.Childrens[:index]
			n.Keys = n.Keys[:index-1]
		} else {
			n.Childrens = append(n.Childrens[:index+1], n.Childrens[index+2:]...)
			n.Keys = append(n.Keys[:index], n.Keys[index+1:]...)
		}
		// add right to left
		leftClone.Values = append(leftClone.Values, right.Values...)
		leftClone.Keys = append(leftClone.Keys, right.Keys...)
	} else {
		leftClone.Keys = append(leftClone.Keys, append([][]byte{n.Keys[index]}, right.Keys...)...)
		// merge childrens
		leftClone.Childrens = append(leftClone.Childrens, right.Childrens...)
		// remove old key
		n.Keys = append(n.Keys[:index], n.Keys[index+1:]...)
		// remove old right node
		n.Childrens = append(n.Childrens[:index+1], n.Childrens[index+2:]...)
		// check size, spilt if over size
		if len(leftClone.Keys) > int(tree.GetNodeMax()) {
			key, left, right := leftClone.split(tree)
			n.insertOnce(key, left, right, tree)
		}
	}
	atomic.StoreInt32(right.IsDirt, 1)
	atomic.StoreInt32(left.IsDirt, 1)
	tree.Nodes[right.GetId()], err = proto.Marshal(right)
	tree.Nodes[left.GetId()], err = proto.Marshal(left)
	tree.Nodes[leftClone.GetId()], err = proto.Marshal(leftClone)
	return leftClone.GetId()
}

func WriteBlock(file *os.File, block proto.Message, blockType string) error {
	blobContent, err := proto.Marshal(block)
	if err != nil {
		return err
	}

	var blobContentLength int32 = int32(len(blobContent))

	var compressedBlob bytes.Buffer
	zlibWriter := zlib.NewWriter(&compressedBlob)
	zlibWriter.Write(blobContent)
	zlibWriter.Close()

	blob := OSMPBF.Blob{}
	blob.ZlibData = compressedBlob.Bytes()
	blob.RawSize = &blobContentLength
	blobBytes, err := proto.Marshal(&blob)
	if err != nil {
		return err
	}

	var blobBytesLength int32 = int32(len(blobBytes))

	blobHeader := OSMPBF.BlobHeader{}
	blobHeader.Type = &blockType
	blobHeader.Datasize = &blobBytesLength
	blobHeaderBytes, err := proto.Marshal(&blobHeader)
	if err != nil {
		return err
	}

	var blobHeaderLength int32 = int32(len(blobHeaderBytes))

	err = binary.Write(file, binary.BigEndian, blobHeaderLength)
	if err != nil {
		return err
	}
	_, err = file.Write(blobHeaderBytes)
	if err != nil {
		return err
	}
	_, err = file.Write(blobBytes)
	if err != nil {
		return err
	}

	return nil
}

func (conn *ProtoBufConn) Call(cmd uint32, args interface{}) (err error) {

	var msg proto.Message
	var buf []byte

	switch m := args.(type) {
	case proto.Message:
		msg = m

		buf, err = proto.Marshal(msg)
		if err != nil {
			return err
		}
	case []byte:
		buf = m
	default:
		return fmt.Errorf("Call args type error %v", args)
	}

	req := &protobuf.Packet{}
	req.Cmd = &cmd
	req.SerializedPacket = buf

	return conn.writeRequest(req)
}

// Request serializes the data (using protobuf), adds the header and sends it to Riak.
func (c *Client) request(req proto.Message, name string) (err error, conn *net.TCPConn) {
	err, conn = c.getConn()
	if err != nil {
		return err, nil
	}
	// Serialize the request using protobuf
	pbmsg, err := proto.Marshal(req)
	if err != nil {
		return err, conn
	}
	// Build message with header: <length:32> <msg_code:8> <pbmsg>
	i := int32(len(pbmsg) + 1)
	msgbuf := []byte{byte(i >> 24), byte(i >> 16), byte(i >> 8), byte(i), messageCodes[name]}
	msgbuf = append(msgbuf, pbmsg...)
	// Send to Riak
	err = c.write(conn, msgbuf)
	// If an error occurred when sending request
	if err != nil {
		// Make sure connection will be released in the end
		defer c.releaseConn(conn)

		var errno syscall.Errno

		// If the error is not recoverable like a broken pipe, close all connections,
		// so next time when getConn() is called it will Connect() again
		if operr, ok := err.(*net.OpError); ok {
			if errno, ok = operr.Err.(syscall.Errno); ok {
				if errno == syscall.EPIPE {
					c.Close()
				}
			}
		}
	}
	return err, conn
}

// usageString returns a description of the amount of space taken up by a body
// with the given contents and a bool indicating overflow.
func (draft *Draft) usageString() (string, bool) {
	var replyToId *uint64
	if draft.inReplyTo != 0 {
		replyToId = proto.Uint64(1)
	}
	var dhPub [32]byte

	msg := &pond.Message{
		Id:               proto.Uint64(0),
		Time:             proto.Int64(1 << 62),
		Body:             []byte(draft.body),
		BodyEncoding:     pond.Message_RAW.Enum(),
		InReplyTo:        replyToId,
		MyNextDh:         dhPub[:],
		Files:            draft.attachments,
		DetachedFiles:    draft.detachments,
		SupportedVersion: proto.Int32(protoVersion),
	}

	serialized, err := proto.Marshal(msg)
	if err != nil {
		panic("error while serialising candidate Message: " + err.Error())
	}

	s := fmt.Sprintf("%d of %d bytes", len(serialized), pond.MaxSerializedMessage)
	return s, len(serialized) > pond.MaxSerializedMessage
}

func (c *context) Call(service, method string, in, out ProtoMessage, opts *CallOptions) error {
	if service == "__go__" {
		if method == "GetNamespace" {
			out.(*basepb.StringProto).Value = proto.String(c.req.Header.Get("X-AppEngine-Current-Namespace"))
			return nil
		}
		if method == "GetDefaultNamespace" {
			out.(*basepb.StringProto).Value = proto.String(c.req.Header.Get("X-AppEngine-Default-Namespace"))
			return nil
		}
	}
	if f, ok := apiOverrides[struct{ service, method string }{service, method}]; ok {
		return f(in, out, opts)
	}
	data, err := proto.Marshal(in)
	if err != nil {
		return err
	}

	requestID := c.req.Header.Get("X-Appengine-Internal-Request-Id")
	res, err := call(service, method, data, requestID)
	if err != nil {
		return err
	}
	return proto.Unmarshal(res, out)
}

func mustMarshal(p interface{}) []byte {
	buf, err := proto.Marshal(p)
	if err != nil {
		panic(err)
	}
	return buf
}

// Heart 测试
func tstHeart() {
	tstfun := "tstHeart"
	util.LogInfo("<<<<<<%s TEST", tstfun)
	syn := &pushproto.Talk{
		Type: pushproto.Talk_HEART.Enum(),
	}

	data, err := proto.Marshal(syn)
	if err != nil {
		util.LogError("%s ERROR:proto marshal error:%s", tstfun, err)
		return
	}

	sb := util.Packdata(data)
	tstErr(tstfun, sb, 1, func(pb *pushproto.Talk) {
		pb_type := pb.GetType()
		if pb_type == pushproto.Talk_HEART {
			util.LogInfo(">>>>>>%s PASS", tstfun)
		} else {
			util.LogError("%s ERROR", tstfun)
		}

	})

}

func mustMarshal(p proto.Message) []byte {
	buf, err := proto.Marshal(p)
	if err != nil {
		panic(err)
	}
	return buf
}

func (conn *ProtoBufConn) WriteObj(value interface{}) error {

	var msg proto.Message

	switch m := value.(type) {
	case proto.Message:
		msg = m
	default:
		return fmt.Errorf("WriteObj value type error %v", value)
	}

	buf, err := proto.Marshal(msg)
	if err != nil {
		return err
	}

	req := &protobuf.Packet{}

	t := reflect.Indirect(reflect.ValueOf(msg)).Type()
	if conn.resultServer == "" {
		req.SetCmd(conn.protocol[t.PkgPath()+"."+t.Name()])
	} else {
		req.SetCmd(conn.protocol[conn.resultServer+"."+t.Name()])
	}
	req.SerializedPacket = buf
	return conn.writeRequest(req)
}