func encodeListTxIndexes(listTx []uint64) []byte {
	b := proto.NewBuffer([]byte{})
	for i := range listTx {
		b.EncodeVarint(listTx[i])
	}
	return b.Bytes()
}

func (trieNode *trieNode) marshal() ([]byte, error) {
	buffer := proto.NewBuffer([]byte{})

	// write value
	err := buffer.EncodeRawBytes(trieNode.value)
	if err != nil {
		return nil, err
	}

	numCryptoHashes := trieNode.getNumChildren()

	//write number of crypto-hashes
	err = buffer.EncodeVarint(uint64(numCryptoHashes))
	if err != nil {
		return nil, err
	}

	if numCryptoHashes == 0 {
		return buffer.Bytes(), nil
	}

	for i, cryptoHash := range trieNode.childrenCryptoHashes {
		//write crypto-hash Index
		err = buffer.EncodeVarint(uint64(i))
		if err != nil {
			return nil, err
		}
		// write crypto-hash
		err = buffer.EncodeRawBytes(cryptoHash)
		if err != nil {
			return nil, err
		}
	}
	return buffer.Bytes(), nil
}

func TestMsg(t *testing.T) {
	recMsg := &OSMsg{Fromu: proto.String("jack"), Tou: proto.String("tom"), Content: proto.String("first")}
	b, err := proto.Marshal(recMsg)
	if checkerr(err) {
		return
	}

	buf := proto.NewBuffer(b)
	err = proto.Unmarshal(b, &msg)
	if checkerr(err) {
		return
	}
	fmt.Println(msg)
	fmt.Println(msg.String())

	err = proto.Unmarshal(buf.Bytes(), &osmsg)
	if checkerr(err) {
		return
	}

	fmt.Println(osmsg)
	fmt.Println(osmsg.String())

	a := []int{1, 2}
	a1 := a[:1]
	fmt.Println(a, a1)
	a1 = append(a1, 3)
	fmt.Println(a, a1)
	a1 = append(a1, 4)
	fmt.Println(a, a1)
	a1[1] = 10
	fmt.Println(a, a1)
}

// BenchmarkVarint64ArrayMixed shows the performance of lots of small messages, each
// containing a small number of large (3, 4, and 5 byte) repeated int64s.
func BenchmarkVarint64ArrayMixed(b *testing.B) {
	for i := uint(1); i <= 1<<5; i <<= 1 {
		dist := genUint64Dist([11]int{0, 0, 0, 4, 6, 4, 0, 0, 0, 0, 0}, int(i))
		// number of sub fields
		for k := uint(1); k <= 1<<10; k <<= 2 {
			msg := &tpb.Message{}
			for m := uint(0); m < k; m++ {
				msg.Children = append(msg.Children, &tpb.Message{
					Key: dist,
				})
			}
			raw, err := proto.Marshal(msg)
			if err != nil {
				b.Error("wrong encode", err)
			}
			b.Run(fmt.Sprintf("Fields%vLen%v", k, i), func(b *testing.B) {
				scratchBuf := proto.NewBuffer(nil)
				b.ResetTimer()
				for k := 0; k < b.N; k++ {
					scratchBuf.SetBuf(raw)
					msgBlackhole.Reset()
					if err := scratchBuf.Unmarshal(msgBlackhole); err != nil {
						b.Error("wrong decode", err)
					}
				}
			})
		}
	}
}

func unmarshalTrieNode(key *trieKey, serializedContent []byte) (*trieNode, error) {
	trieNode := newTrieNode(key, nil, false)
	buffer := proto.NewBuffer(serializedContent)
	value, err := buffer.DecodeRawBytes(false)
	if err != nil {
		return nil, err
	}
	trieNode.value = value

	numCryptoHashes, err := buffer.DecodeVarint()
	if err != nil {
		return nil, err
	}

	for i := uint64(0); i < numCryptoHashes; i++ {
		index, err := buffer.DecodeVarint()
		if err != nil {
			return nil, err
		}
		cryptoHash, err := buffer.DecodeRawBytes(false)
		if err != nil {
			return nil, err
		}
		trieNode.childrenCryptoHashes[int(index)] = cryptoHash
	}
	return trieNode, nil
}

func (bucketNode *bucketNode) marshal() []byte {
	buffer := proto.NewBuffer([]byte{})
	for i := 0; i < conf.getMaxGroupingAtEachLevel(); i++ {
		buffer.EncodeRawBytes(bucketNode.childrenCryptoHash[i])
	}
	return buffer.Bytes()
}

func unmarshalTrieNode(key *trieKey, serializedContent []byte) (*trieNode, error) {
	stateTrieLogger.Debug("key = [%s], len(serializedContent) = %d", key, len(serializedContent))
	trieNode := newTrieNode(key, nil, false)
	buffer := proto.NewBuffer(serializedContent)
	trieNode.value = unmarshalTrieNodeValueFromBuffer(buffer)

	numCryptoHashes, err := buffer.DecodeVarint()
	stateTrieLogger.Debug("numCryptoHashes = [%d]", numCryptoHashes)
	if err != nil {
		return nil, err
	}
	for i := uint64(0); i < numCryptoHashes; i++ {
		index, err := buffer.DecodeVarint()
		if err != nil {
			return nil, err
		}
		cryptoHash, err := buffer.DecodeRawBytes(false)
		if err != nil {
			return nil, err
		}
		trieNode.childrenCryptoHashes[int(index)] = cryptoHash
	}
	stateTrieLogger.Debug("unmarshalled trieNode = [%s]", trieNode)
	return trieNode, nil
}

func (c *codec) Encoder(w io.Writer) mc.Encoder {
	return &encoder{
		w:   w,
		buf: proto.NewBuffer(nil),
		c:   c,
	}
}

func (s *Server) writeResponse(c *Client, cmd int, seq uint32, pb proto.Message) error {
	var err error

	c.sendlock.Lock()
	defer c.sendlock.Unlock()

	// cmd
	binary.LittleEndian.PutUint16(c.wb[0:2], uint16(cmd+1))
	// seq
	binary.LittleEndian.PutUint32(c.wb[4:8], seq)
	logger.Println("send header:", cmd+1, seq)
	// pb
	if pb != nil {
		buf := proto.NewBuffer(c.wb[0:8])
		buf.Marshal(pb)
		respb := buf.Bytes()
		// len
		binary.LittleEndian.PutUint16(respb[2:4], uint16(len(respb)))
		_, err = c.w.Write(respb)
	} else {
		// len
		binary.LittleEndian.PutUint16(c.wb[2:4], uint16(HeaderSize))
		_, err = c.w.Write(c.wb[0:8])
	}
	return err
}

func (s *Server) NotifyClient(c *Client, cmd int, pb proto.Message, cb NotifyCallback) error {
	var err error

	seq := c.addPendingRequest(cb)

	c.sendlock.Lock()
	defer c.sendlock.Unlock()

	// cmd
	binary.LittleEndian.PutUint16(c.wb[0:2], uint16(cmd))
	// seq
	binary.LittleEndian.PutUint32(c.wb[4:8], seq)
	// pb
	if pb != nil {
		buf := proto.NewBuffer(c.wb[0:8])
		buf.Marshal(pb)
		respb := buf.Bytes()
		// len
		binary.LittleEndian.PutUint16(respb[2:4], uint16(len(respb)))
		_, err = c.w.Write(respb)
	} else {
		// len
		binary.LittleEndian.PutUint16(c.wb[2:4], uint16(HeaderSize))
		_, err = c.w.Write(c.wb[0:8])
	}

	return err
}

// dataAsString returns the given byte array as a string; handles detecting
// protocol buffers.
func dataAsString(data []byte) string {
	dataString := string(data)
	if !strings.HasPrefix(dataString, "{") {
		dataString = "\n" + hex.Dump(data)
		proto.NewBuffer(make([]byte, 0, 1024)).DebugPrint("decoded object", data)
	}
	return dataString
}

func unmarshalTrieNodeValue(serializedContent []byte) []byte {
	buffer := proto.NewBuffer(serializedContent)
	value, err := buffer.DecodeRawBytes(false)
	if err != nil {
		panic(fmt.Errorf("This error is not excpected: %s", err))
	}
	return value
}

func (self *Shard) Write(database string, series []*protocol.Series) error {
	self.closeLock.RLock()
	defer self.closeLock.RUnlock()
	if self.closed {
		return fmt.Errorf("Shard is closed")
	}

	wb := make([]storage.Write, 0)

	for _, s := range series {
		if len(s.Points) == 0 {
			return errors.New("Unable to write no data. Series was nil or had no points.")
		}
		if len(s.FieldIds) == 0 {
			return errors.New("Unable to write points without fields")
		}

		count := 0
		for fieldIndex, id := range s.FieldIds {
			for _, point := range s.Points {
				// keyBuffer and dataBuffer have to be recreated since we are
				// batching the writes, otherwise new writes will override the
				// old writes that are still in memory
				dataBuffer := proto.NewBuffer(nil)
				var err error

				sk := newStorageKey(id, point.GetTimestamp(), point.GetSequenceNumber())
				if point.Values[fieldIndex].GetIsNull() {
					wb = append(wb, storage.Write{Key: sk.bytes(), Value: nil})
					goto check
				}

				err = dataBuffer.Marshal(point.Values[fieldIndex])
				if err != nil {
					return err
				}
				wb = append(wb, storage.Write{Key: sk.bytes(), Value: dataBuffer.Bytes()})
			check:
				count++
				if count >= self.writeBatchSize {
					err = self.db.BatchPut(wb)
					if err != nil {
						return err
					}
					count = 0
					wb = make([]storage.Write, 0, self.writeBatchSize)
				}
			}
		}
	}

	return self.db.BatchPut(wb)
}

func decodeBlockNumTxIndex(bytes []byte) (blockNum uint64, txIndex uint64, err error) {
	b := proto.NewBuffer(bytes)
	blockNum, err = b.DecodeVarint()
	if err != nil {
		return
	}
	txIndex, err = b.DecodeVarint()
	if err != nil {
		return
	}
	return
}

// Marshal serializes a `TxReadWriteSet`
func (txRW *TxReadWriteSet) Marshal() ([]byte, error) {
	buf := proto.NewBuffer(nil)
	var err error
	if err = buf.EncodeVarint(uint64(len(txRW.NsRWs))); err != nil {
		return nil, err
	}
	for i := 0; i < len(txRW.NsRWs); i++ {
		if err = txRW.NsRWs[i].Marshal(buf); err != nil {
			return nil, err
		}
	}
	return buf.Bytes(), nil
}

func writeFromPointers(queryid string, f io.Writer, pointers []resultPointer) error {
	stateMu.RLock()
	s := state[queryid]
	stateMu.RUnlock()
	firstPathRank := s.FirstPathRank

	s.tempFilesMu.Lock()
	defer s.tempFilesMu.Unlock()

	if _, err := f.Write([]byte("[")); err != nil {
		return err
	}
	var msg pb.SearchReply
	buf := proto.NewBuffer(nil)
	for idx, pointer := range pointers {
		src := s.perBackend[pointer.backendidx].tempFile
		if _, err := src.Seek(pointer.offset, os.SEEK_SET); err != nil {
			return err
		}
		// TODO: Avoid the allocations by using a slice and only allocate a new buffer when pointer.length > cap(rdbuf)
		rdbuf := make([]byte, pointer.length)
		if _, err := src.Read(rdbuf); err != nil {
			return err
		}
		if idx > 0 {
			if _, err := f.Write([]byte(",")); err != nil {
				return err
			}
		}
		buf.SetBuf(rdbuf)
		msg.Reset()
		if err := buf.Unmarshal(&msg); err != nil {
			return err
		}
		if msg.Type != pb.SearchReply_MATCH {
			return fmt.Errorf("Expected to find a pb.SearchReply_MATCH, instead got %d", msg.Type)
		}
		match := msg.Match
		// We need to fix the ranking here because we persist raw results from
		// the dcs-source-backend in queryBackend(), but then modify the
		// ranking in storeResult().
		match.Ranking = match.Pathrank + ((firstPathRank * 0.1) * match.Ranking)
		if err := WriteMatchJSON(match, f); err != nil {
			return err
		}
	}
	if _, err := f.Write([]byte("]\n")); err != nil {
		return err
	}
	return nil
}

func unmarshalBucketNode(bucketKey *bucketKey, serializedBytes []byte) *bucketNode {
	bucketNode := newBucketNode(bucketKey)
	buffer := proto.NewBuffer(serializedBytes)
	for i := 0; i < conf.getMaxGroupingAtEachLevel(); i++ {
		childCryptoHash, err := buffer.DecodeRawBytes(false)
		if err != nil {
			panic(fmt.Errorf("this error should not occur: %s", err))
		}
		if !util.IsNil(childCryptoHash) {
			bucketNode.childrenCryptoHash[i] = childCryptoHash
		}
	}
	return bucketNode
}

func (i *checkpointInfo) marshal() ([]byte, error) {
	buffer := proto.NewBuffer([]byte{})
	var err error
	if err = buffer.EncodeVarint(uint64(i.latestFileChunkSuffixNum)); err != nil {
		return nil, err
	}
	if err = buffer.EncodeVarint(uint64(i.latestFileChunksize)); err != nil {
		return nil, err
	}
	if err = buffer.EncodeVarint(i.lastBlockNumber); err != nil {
		return nil, err
	}
	return buffer.Bytes(), nil
}

// Marshal serializes the StateDelta
func (stateDelta *StateDelta) Marshal() (b []byte) {
	buffer := proto.NewBuffer([]byte{})
	err := buffer.EncodeVarint(uint64(len(stateDelta.chaincodeStateDeltas)))
	if err != nil {
		// in protobuf code the error return is always nil
		panic(fmt.Errorf("This error should not occure: %s", err))
	}
	for chaincodeID, chaincodeStateDelta := range stateDelta.chaincodeStateDeltas {
		buffer.EncodeStringBytes(chaincodeID)
		chaincodeStateDelta.marshal(buffer)
	}
	b = buffer.Bytes()
	return
}

func processMessage(msg []byte) [][]byte {
	buf := pb.NewBuffer(msg)
	payloads := make([][]byte, 0)

	for {
		hbytes, err := buf.DecodeRawBytes(true)
		if err != nil {
			break
		}

		payloads = append(payloads, hbytes)
	}

	return payloads
}