// Serialize converts this Scan into a serialized protobuf message ready
// to be sent to an HBase node.
func (s *Scan) Serialize() ([]byte, error) {
	scan := &pb.ScanRequest{
		Region:       s.regionSpecifier(),
		CloseScanner: &s.closeScanner,
		NumberOfRows: &s.numberOfRows,
	}
	if s.scannerID != math.MaxUint64 {
		scan.ScannerId = &s.scannerID
		return proto.Marshal(scan)
	}
	scan.Scan = &pb.Scan{
		Column:    familiesToColumn(s.families),
		StartRow:  s.startRow,
		StopRow:   s.stopRow,
		TimeRange: &pb.TimeRange{},
	}
	if s.maxVersions != DefaultMaxVersions {
		scan.Scan.MaxVersions = &s.maxVersions
	}
	if s.fromTimestamp != MinTimestamp {
		scan.Scan.TimeRange.From = &s.fromTimestamp
	}
	if s.toTimestamp != MaxTimestamp {
		scan.Scan.TimeRange.To = &s.toTimestamp
	}

	if s.filters != nil {
		pbFilter, err := s.filters.ConstructPBFilter()
		if err != nil {
			return nil, err
		}
		scan.Scan.Filter = pbFilter
	}
	return proto.Marshal(scan)
}

//挑战其他玩家
func (this *Player) ChallengePlayer(type_ int32, other_role_id int64) {
	result4C := new(protocol.StageBase_ChallengePlayerResult)
	if type_ == 1 { //挂机挑战其他玩家
		result := this.Guaji_Stage.IsCanPk(this.PlayerId, other_role_id)
		result4C.IsCanChange = &result

		if !result { //不能挑战直接返回
			encObj, _ := proto.Marshal(result4C)
			SendPackage(*this.conn, 1111, encObj)
		}

		Heros := make(map[int32]*HeroStruct)
		if value, ok := word.players[other_role_id]; ok {
			for k, v := range value.Heros { //遍历对应玩家的所有英雄列表
				if v.Hero_Info.Pos_stage > 0 { //关卡上阵英雄
					Heros[k] = v
				}
			}
			Game_HeroStruct := this.GetHeroStruct(Heros)
			result4C.Team_2 = Game_HeroStruct
		}

		encObj, _ := proto.Marshal(result4C)
		SendPackage(*this.conn, 1111, encObj)
	}
}

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

	this.mutex.Lock()
	req.Id = proto.Uint64(this.next)
	this.next++
	this.pending[*req.Id] = &rr
	this.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(this.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
}

//TestUpgradeNonExistChaincode tests upgrade non exist chaincode
func TestUpgradeNonExistChaincode(t *testing.T) {
	initialize()

	scc := new(LifeCycleSysCC)
	stub := shim.NewMockStub("lccc", scc)

	cds, err := constructDeploymentSpec("example02", "github.com/hyperledger/fabric/examples/chaincode/go/chaincode_example02", [][]byte{[]byte("init"), []byte("a"), []byte("100"), []byte("b"), []byte("200")})
	var b []byte
	if b, err = proto.Marshal(cds); err != nil || b == nil {
		t.Fatalf("Marshal DeploymentSpec failed")
	}

	args := [][]byte{[]byte(DEPLOY), []byte("test"), b}
	if _, err := stub.MockInvoke("1", args); err != nil {
		t.Fatalf("Deploy chaincode error: %v", err)
	}

	newCds, err := constructDeploymentSpec("example03", "github.com/hyperledger/fabric/examples/chaincode/go/chaincode_example02", [][]byte{[]byte("init"), []byte("a"), []byte("100"), []byte("b"), []byte("200")})
	var newb []byte
	if newb, err = proto.Marshal(newCds); err != nil || newb == nil {
		t.Fatalf("Marshal DeploymentSpec failed")
	}

	args = [][]byte{[]byte(UPGRADE), []byte("test"), newb}
	_, err = stub.MockInvoke("1", args)
	if _, ok := err.(NotFoundErr); !ok {
		t.FailNow()
	}
}

func (c *serverCodec) WriteResponse(r *rpc.Response, result interface{}) error {
	rsp := &PbRpcResponse{Id: &r.Seq}

	if r.Error != "" {
		rsp.Error = &r.Error
	} else {
		protoResult, ok := result.(proto.Message)
		if !ok {
			return errors.New("Invalid type given")
		}
		data, err := proto.Marshal(protoResult)
		if err != nil {
			return err
		}
		rsp.Result = data
	}

	data, err := proto.Marshal(rsp)
	if err != nil {
		return err
	}
	_, err = WriteRpc(c.rwc, data)

	return err
}

func TestLedgerReadWrite(t *testing.T) {
	localConf := localconfig.Load()
	localConf.General.OrdererType = provisional.ConsensusTypeSbft
	genesis := provisional.New(localConf).GenesisBlock()
	_, rl := ramledger.New(10, genesis)
	b := Backend{ledger: rl}

	header := []byte("header")
	e1 := &cb.Envelope{Payload: []byte("data1")}
	e2 := &cb.Envelope{Payload: []byte("data2")}
	ebytes1, _ := proto.Marshal(e1)
	ebytes2, _ := proto.Marshal(e2)
	data := [][]byte{ebytes1, ebytes2}
	sgns := make(map[uint64][]byte)
	sgns[uint64(1)] = []byte("sgn1")
	sgns[uint64(22)] = []byte("sgn22")
	batch := simplebft.Batch{Header: header, Payloads: data, Signatures: sgns}

	b.Deliver(&batch)
	batch2 := b.LastBatch()

	if !reflect.DeepEqual(batch, *batch2) {
		t.Errorf("The wrong batch was returned by LastBatch after Deliver: %v (original was: %v)", batch2, &batch)
	}
}

// Returns a valid protobuf collection in bytes and the related jobs
func GetTestProtobufCollectionBody(userId uint32, numberOfPayloads int) (*TrackBodyCollection, []*EventAction) {
	collection, wrongCollection, incompleteCollection := GetTestCollectionPairs(userId, numberOfPayloads)
	collectionBytestream, _ := proto.Marshal(collection)
	disturbedCollectionBytestream, _ := proto.Marshal(wrongCollection)
	incompleteCollectionByteStream, _ := proto.Marshal(incompleteCollection)
	return &TrackBodyCollection{collectionBytestream, disturbedCollectionBytestream, incompleteCollectionByteStream}, GetJobsFromCollection(collection)
}

// 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
}

// 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
}

func TestVerifyHostAttestation_rootHost(t *testing.T) {
	domain := generateDomain(t)
	policyKey, policyCert := domain.Keys, domain.Keys.Cert
	hostKey, hostAtt := generateAttestation(t, policyKey, hostName)
	programKey, programAtt := generateAttestation(t, hostKey, programName)
	rawEnd, err := proto.Marshal(hostAtt)
	if err != nil {
		t.Fatal("Error serializing attestation.")
	}
	programAtt.SerializedEndorsements = [][]byte{rawEnd}
	rawAtt, err := proto.Marshal(programAtt)
	if err != nil {
		t.Fatal("Error serializing attestation.")
	}
	certPool := x509.NewCertPool()
	certPool.AddCert(policyCert)
	speaker, key, prog, err := VerifyHostAttestation(rawAtt, domain, certPool)
	if err != nil {
		t.Fatal("Test attesation failed verification checks.", err)
	}
	if !programName.Identical(prog) {
		t.Fatal("Attestation program name not identical to expected program name.")
	}
	if !programKey.SigningKey.ToPrincipal().Identical(key) {
		t.Fatal("Attestation program key not identical to expected program key.")
	}
	if !hostKey.SigningKey.ToPrincipal().Identical(speaker) {
		t.Fatal("Attestation host key not identical to expected host key.")
	}
}

func Action1Handler(user *User, data []byte) {

	// request body unmarshaling
	req := new(gs_protocol.ReqAction1)
	err := proto.Unmarshal(data, req)
	gs.CheckError(err)

	// TODO create business logic for Action1 Type
	if DEBUG {
		gs.Log("Action1 userID : ", gs.Itoa64(req.GetUserID()))
	}

	// broadcast message
	notifyMsg := new(gs_protocol.NotifyAction1Msg)
	notifyMsg.UserID = proto.Int64(user.userID)
	msg, err := proto.Marshal(notifyMsg)
	gs.CheckError(err)

	user.SendToAll(NewMessage(user.userID, gs_protocol.Type_NotifyAction1, msg))

	// response body marshaling
	res := new(gs_protocol.ResAction1)
	res.UserID = proto.Int64(user.userID)
	res.Result = proto.Int32(1) // is success?
	msg, err = proto.Marshal(res)
	gs.CheckError(err)
	user.Push(NewMessage(user.userID, gs_protocol.Type_DefinedAction1, msg))
}

func writeEvent(ev *Event, writer io.Writer) error {
	if len(ev.raw) > 0 {
		_, err := writer.Write(ev.raw)
		return err
	}
	var buf bytes.Buffer
	dataBuf, err := proto.Marshal(ev.Msg)
	if nil != err {
		return err
	}
	msgtype := proto.MessageName(ev.Msg)
	ev.MsgType = &msgtype
	headbuf, _ := proto.Marshal(ev)
	headerLen := uint32(len(headbuf))

	// length = msglength(3byte) + headerlen(1byte)
	length := uint32(len(dataBuf))
	length = (length << 8) + headerLen
	buf.Write(MAGIC_EVENT_HEADER)
	binary.Write(&buf, binary.LittleEndian, length)

	buf.Write(headbuf)
	buf.Write(dataBuf)
	_, err = buf.WriteTo(writer)

	return err
}

func (op *obcBatch) sendBatch() error {
	op.stopBatchTimer()
	// assemble new Request message
	txs := make([]*pb.Transaction, len(op.batchStore))
	var i int
	for d, req := range op.batchStore {
		txs[i] = &pb.Transaction{}
		err := proto.Unmarshal(req.Payload, txs[i])
		if err != nil {
			err = fmt.Errorf("Unable to unpack payload of request %d", i)
			logger.Error("%s", err)
			return err
		}
		i++
		delete(op.batchStore, d) // clean up
	}
	tb := &pb.TransactionBlock{Transactions: txs}
	tbPacked, err := proto.Marshal(tb)
	if err != nil {
		err = fmt.Errorf("Unable to pack transaction block for new batch request")
		logger.Error("%s", err)
		return err
	}
	// process internally
	op.pbft.request(tbPacked)
	// broadcast
	batchReq := &Request{Payload: tbPacked}
	msg := &Message{&Message_Request{batchReq}}
	msgRaw, _ := proto.Marshal(msg)
	op.broadcast(msgRaw)

	return nil
}

// NewChaincodeDeployTransaction is used to deploy chaincode.
func NewChaincodeDeployTransaction(chaincodeDeploymentSpec *ChaincodeDeploymentSpec, uuid string) (*Transaction, error) {
	transaction := new(Transaction)
	transaction.Type = Transaction_CHAINCODE_DEPLOY
	transaction.Uuid = uuid
	transaction.Timestamp = util.CreateUtcTimestamp()
	cID := chaincodeDeploymentSpec.ChaincodeSpec.GetChaincodeID()
	if cID != nil {
		data, err := proto.Marshal(cID)
		if err != nil {
			return nil, fmt.Errorf("Could not marshal chaincode : %s", err)
		}
		transaction.ChaincodeID = data
	}
	//if chaincodeDeploymentSpec.ChaincodeSpec.GetCtorMsg() != nil {
	//	transaction.Function = chaincodeDeploymentSpec.ChaincodeSpec.GetCtorMsg().Function
	//	transaction.Args = chaincodeDeploymentSpec.ChaincodeSpec.GetCtorMsg().Args
	//}
	data, err := proto.Marshal(chaincodeDeploymentSpec)
	if err != nil {
		logger.Errorf("Error mashalling payload for chaincode deployment: %s", err)
		return nil, fmt.Errorf("Could not marshal payload for chaincode deployment: %s", err)
	}
	transaction.Payload = data
	return transaction, nil
}

func TestMultiPart(t *testing.T) {
	stream := &fakeStream{
		recvPackets: make(chan shanPacket, 5),
		sendPackets: make(chan shanPacket, 5),
	}

	controller := setupTestController(stream)

	subHeader := &Spotify.Header{
		Uri: proto.String("hm://searchview/km/v2/search/Future"),
	}
	subHeaderData, _ := proto.Marshal(subHeader)

	header := &Spotify.Header{
		Uri:         proto.String("hm://searchview/km/v2/search/Future"),
		ContentType: proto.String("application/json; charset=UTF-8"),
		StatusCode:  proto.Int32(200),
	}
	body := []byte("{searchResults: {tracks: [], albums: [], tracks: []}}")

	headerData, _ := proto.Marshal(header)
	seq := []byte{0, 0, 0, 1}

	p0, _ := encodeMercuryHead([]byte{0, 0, 0, 0}, 1, 1)
	binary.Write(p0, binary.BigEndian, uint16(len(subHeaderData)))
	p0.Write(subHeaderData)

	p1, _ := encodeMercuryHead(seq, 1, 0)
	binary.Write(p1, binary.BigEndian, uint16(len(headerData)))
	p1.Write(headerData)

	p2, _ := encodeMercuryHead(seq, 1, 1)
	binary.Write(p2, binary.BigEndian, uint16(len(body)))
	p2.Write(body)

	didRecieveCallback := false
	controller.session.mercurySendRequest(mercuryRequest{
		method:  "SEND",
		uri:     "hm://searchview/km/v2/search/Future",
		payload: [][]byte{},
	}, func(res mercuryResponse) {
		didRecieveCallback = true
		if string(res.payload[0]) != string(body) {
			t.Errorf("bad body received")
		}
	})

	stream.recvPackets <- shanPacket{cmd: 0xb2, buf: p0.Bytes()}
	stream.recvPackets <- shanPacket{cmd: 0xb2, buf: p1.Bytes()}
	stream.recvPackets <- shanPacket{cmd: 0xb2, buf: p2.Bytes()}

	controller.session.poll()
	controller.session.poll()
	controller.session.poll()

	if !didRecieveCallback {
		t.Errorf("never received callback")
	}

}

func getfetchMsg(key [sha256.Size]byte) []byte {

	msg := new(fsMsgs.NetworkMessage)
	msg.Proto = proto.Uint32(2)
	appMsg := new(fsMsgs.AppMessage)
	fsMsg := new(fsMsgs.AppMessage_FileSystemMessage)
	command := fsMsgs.AppMessage_Command(fsMsgs.AppMessage_Command_value["FETCH"])
	fsMsg.Cmd = &command
	fetchMsg := new(fsMsgs.FetchMessage)
	fetchMsg.Key = proto.String(string(key[:sha256.Size]))
	fsMsg.Fmsg = fetchMsg
	appMsg.Msg = fsMsg

	appdata, err := proto.Marshal(appMsg)
	if err != nil {
		log.Fatal("marshaling error: ", err)
	}
	msg.Msg = proto.String(string(appdata))

	data, err := proto.Marshal(msg)
	if err != nil {
		log.Fatal("marshaling error: ", err)
	}

	return data
}

func (c *mercuryCodec) Write(m *codec.Message, b interface{}) error {
	switch m.Type {
	case codec.Request:
		data, err := proto.Marshal(b.(proto.Message))
		if err != nil {
			return err
		}
		c.rwc.Write(data)
		m.Header["Content-Encoding"] = "request"
		m.Header["Service"] = m.Target
		m.Header["Endpoint"] = m.Method
	case codec.Response:
		m.Header["Content-Encoding"] = "response"
		data, err := proto.Marshal(b.(proto.Message))
		if err != nil {
			return err
		}
		c.rwc.Write(data)
	case codec.Publication:
		data, err := proto.Marshal(b.(proto.Message))
		if err != nil {
			return err
		}
		c.rwc.Write(data)
	default:
		return fmt.Errorf("Unrecognised message type: %v", m.Type)
	}
	return nil
}

func getstoreMsg(key [sha256.Size]byte, document []byte) []byte {

	msg := new(fsMsgs.NetworkMessage)
	msg.Proto = proto.Uint32(2)
	appMsg := new(fsMsgs.AppMessage)
	fsMsg := new(fsMsgs.AppMessage_FileSystemMessage)
	command := fsMsgs.AppMessage_Command(fsMsgs.AppMessage_Command_value["STORE"])
	fsMsg.Cmd = &command
	storeMsg := new(fsMsgs.StoreMessage)
	storeMsg.Key = proto.String(string(key[:sha256.Size]))
	storeMsg.Document = proto.String(string(document))
	fsMsg.Smsg = storeMsg
	appMsg.Msg = fsMsg

	appdata, err := proto.Marshal(appMsg)
	if err != nil {
		log.Fatal("marshaling error: ", err)
	}
	msg.Msg = proto.String(string(appdata))

	data, err := proto.Marshal(msg)
	if err != nil {
		log.Fatal("marshaling error: ", err)
	}

	return data
}

//TestUpgrade tests the upgrade function
func TestUpgrade(t *testing.T) {
	initialize()

	scc := new(LifeCycleSysCC)
	stub := shim.NewMockStub("lccc", scc)

	cds, err := constructDeploymentSpec("example02", "github.com/hyperledger/fabric/examples/chaincode/go/chaincode_example02", [][]byte{[]byte("init"), []byte("a"), []byte("100"), []byte("b"), []byte("200")})
	var b []byte
	if b, err = proto.Marshal(cds); err != nil || b == nil {
		t.Fatalf("Marshal DeploymentSpec failed")
	}

	args := [][]byte{[]byte(DEPLOY), []byte("test"), b}
	if _, err := stub.MockInvoke("1", args); err != nil {
		t.Fatalf("Deploy chaincode error: %v", err)
	}

	newCds, err := constructDeploymentSpec("example02", "github.com/hyperledger/fabric/examples/chaincode/go/chaincode_example02", [][]byte{[]byte("init"), []byte("a"), []byte("100"), []byte("b"), []byte("200")})
	var newb []byte
	if newb, err = proto.Marshal(newCds); err != nil || newb == nil {
		t.Fatalf("Marshal DeploymentSpec failed")
	}

	args = [][]byte{[]byte(UPGRADE), []byte("test"), newb}
	version, err := stub.MockInvoke("1", args)
	if err != nil {
		t.Fatalf("Upgrade chaincode error: %v", err)
	}

	expectVer := "1"
	newVer := string(version)
	if newVer != expectVer {
		t.Fatalf("Upgrade chaincode version error, expected %s, got %s", expectVer, newVer)
	}
}