本文旨在通過介紹一個(gè)接口改造需求的實(shí)現(xiàn)過程聪轿，分享下筆者在讀捌斧、改Fabric源碼中積累的一點(diǎn)心得戈稿，偏頗之處西土，歡迎指正
需求：
拓展chaincode查詢歷史數(shù)據(jù)接口功能，增加分頁功能
準(zhǔn)備：
1鞍盗、從Fabric fork一個(gè)自己的版本 (我選擇的是 Fabric v1.2.0)
2需了、本地git clone
3、簡單瞄一下源碼
項(xiàng)目結(jié)構(gòu)還算清晰般甲，其中msp肋乍、orderer、peer目錄可也理解為對應(yīng)模塊的入口敷存，且與cli命令一一對應(yīng)住拭，比如說channel 命令 ，對比官網(wǎng)peer channel命令和源碼peer/channel下的文件：
每條命令映射到一個(gè)文件历帚，如何實(shí)現(xiàn)的滔岳？瞄一下 peer/main.go
import(
...
"github.com/spf13/cobra"
...
)
cobra是一個(gè)用來生成CLI的強(qiáng)大工具，參見官網(wǎng) https://github.com/spf13/cobra
找到入口挽牢，就可以在代碼中完整的跟蹤一個(gè)命令的執(zhí)行過程
定位：
需求是擴(kuò)展chaincode接口中的歷史數(shù)據(jù)查詢接口谱煤，增加分頁功能，直接定位到接口文件：core/chaincode/shim/interfaces.go
怎么定位禽拔，最簡單的方法就是IDE中全文搜chaincode中常用方法GetArgs()
找到歷史數(shù)據(jù)查詢接口：
// GetHistoryForKey returns a history of key values across time.
// For each historic key update, the historic value and associated
// transaction id and timestamp are returned. The timestamp is the
// timestamp provided by the client in the proposal header.
// GetHistoryForKey requires peer configuration
// core.ledger.history.enableHistoryDatabase to be true.
// The query is NOT re-executed during validation phase, phantom reads are
// not detected. That is, other committed transactions may have updated
// the key concurrently, impacting the result set, and this would not be
// detected at validation/commit time. Applications susceptible to this
// should therefore not use GetHistoryForKey as part of transactions that
// update ledger, and should limit use to read-only chaincode operations.
GetHistoryForKey(key string) (HistoryQueryIteratorInterface, error)
可以通過上面注釋看到該函數(shù)的返回?cái)?shù)據(jù)格式刘离、配置室叉、使用場景說明。
另外注意Fabric中的“歷史數(shù)據(jù)”記錄的是并不是所有數(shù)據(jù)的操作記錄硫惕，而只是對"world state"即“世界狀態(tài)”中的key-value數(shù)據(jù)新增或變化進(jìn)行記錄茧痕，歷史數(shù)據(jù)庫默認(rèn)用的是leveldb，所以所有歷史記錄也是key-value數(shù)據(jù)恼除，且value值為空踪旷，僅僅只有key，格式為ns?key?blockNo?tranNo豁辉，其中的問號指代分隔符令野，真正的分隔符為[]byte{0x00}。
這里可能會有點(diǎn)不太好理解徽级，為什么歷史記錄只用key气破，value為空，兩個(gè)原因：
第一餐抢，“歷史數(shù)據(jù)”存的是key的create现使、update操作對應(yīng)的blockID、transactionID旷痕，不復(fù)雜朴下；
第二，leveldb適合“隨機(jī)寫苦蒿、順序讀/寫”殴胧，其中的順序讀指的是按字符大小順序，ns?key?blockNo?tranNo存儲剛好滿足范圍查詢時(shí)的順序讀特性佩迟。
調(diào)用流程
前面定位了接口位置团滥，真正要做改造優(yōu)化，還需要知道整個(gè)接口的實(shí)現(xiàn)過程报强，也就是調(diào)用流程灸姊。
上面是筆者總結(jié)的一個(gè)簡單的chaincode接口調(diào)用peer具體實(shí)現(xiàn)的過程，通信采用protobuf+gRPC（沒有接觸過的同學(xué)建議先了解下）秉溉，client相關(guān)函數(shù)主要在core/chaincode/shim的interfaces.go力惯、handler.go、chaincode.go召嘶，server相關(guān)函數(shù)主要在core/chaincode下面的handler.go父晶、chaincode_support.go。
注意兩點(diǎn)：
一個(gè)是gRPC server端的注冊
protobuf文件protos/peer/chaincode_shim.proto弄跌，對應(yīng)的go文件即同目錄下的同名.go文件甲喝，點(diǎn)擊查看chaincode_shim.proto文件
// Interface that provides support to chaincode execution. ChaincodeContext
// provides the context necessary for the server to respond appropriately.
service ChaincodeSupport {
rpc Register(stream ChaincodeMessage) returns (stream ChaincodeMessage) {}
}
發(fā)現(xiàn)只聲明了一個(gè)函數(shù)，且客戶端服務(wù)端都使用stream通信铛只，
該函數(shù)服務(wù)端實(shí)現(xiàn)在chaincode_support.go下：
// Register the bidi stream entry point called by chaincode to register with the Peer.
func (cs *ChaincodeSupport) Register(stream pb.ChaincodeSupport_RegisterServer) error {
return cs.HandleChaincodeStream(stream.Context(), stream)
}
可以追蹤HandleChaincodeStream()方法埠胖，
--->handler.ProcessStream()
--->handler.handleMessage()
--->handler.handleMessageCreatedState() or handler.handleMessageReadyState()
以后者為例
func (h *Handler) handleMessageReadyState(msg *pb.ChaincodeMessage) error {
switch msg.Type {
case pb.ChaincodeMessage_COMPLETED, pb.ChaincodeMessage_ERROR:
h.Notify(msg)
case pb.ChaincodeMessage_PUT_STATE:
go h.HandleTransaction(msg, h.HandlePutState)
case pb.ChaincodeMessage_DEL_STATE:
go h.HandleTransaction(msg, h.HandleDelState)
case pb.ChaincodeMessage_INVOKE_CHAINCODE:
go h.HandleTransaction(msg, h.HandleInvokeChaincode)
case pb.ChaincodeMessage_GET_STATE:
go h.HandleTransaction(msg, h.HandleGetState)
case pb.ChaincodeMessage_GET_STATE_BY_RANGE:
go h.HandleTransaction(msg, h.HandleGetStateByRange)
case pb.ChaincodeMessage_GET_QUERY_RESULT:
go h.HandleTransaction(msg, h.HandleGetQueryResult)
case pb.ChaincodeMessage_GET_HISTORY_FOR_KEY:
go h.HandleTransaction(msg, h.HandleGetHistoryForKey)
case pb.ChaincodeMessage_QUERY_STATE_NEXT:
go h.HandleTransaction(msg, h.HandleQueryStateNext)
case pb.ChaincodeMessage_QUERY_STATE_CLOSE:
go h.HandleTransaction(msg, h.HandleQueryStateClose)
default:
return fmt.Errorf("[%s] Fabric side handler cannot handle message (%s) while in ready state", msg.Txid, msg.Type)
}
return nil
}
即可定位到server端的具體實(shí)現(xiàn)方法糠溜。
一個(gè)是Client端的初始化
入口是chaincode.go 下的start()，即每個(gè)合約文件的main方法中都會調(diào)用的方法直撤，由上圖所述非竿，追蹤到userChaincodeStreamGetter()，其中的：
...
// Establish connection with validating peer
clientConn, err := newPeerClientConnection()
...
chaincodeSupportClient := pb.NewChaincodeSupportClient(clientConn)
// Establish stream with validating peer
stream, err := chaincodeSupportClient.Register(context.Background())
...
即實(shí)現(xiàn)gRPC Client端的初始化谋竖，并調(diào)用pb文件中聲明的唯一方法红柱，建立跟peer節(jié)點(diǎn)注冊的server端的通信。
OK圈盔，大體的調(diào)用流程搞明白，再聚焦到GetHistoryForKey()的實(shí)現(xiàn)悄雅，通過上面的說明驱敲，快速定位到corechincode/handler.go 中的 HandleGetHistoryForKey()方法中的
historyIter, err := txContext.HistoryQueryExecutor.GetHistoryForKey(chaincodeName, getHistoryForKey.Key)
注意其中的historyleveldb_test.go，改造后的代碼可以先在test文件中驗(yàn)證览徒，前提是代碼執(zhí)行本地安裝了docker狈定。
回到代碼
// GetHistoryForKey implements method in interface `ledger.HistoryQueryExecutor`
func (q *LevelHistoryDBQueryExecutor) GetHistoryForKey(namespace string, key string) (commonledger.ResultsIterator, error) {
if ledgerconfig.IsHistoryDBEnabled() == false {
return nil, errors.New("History tracking not enabled - historyDatabase is false")
}
var compositeStartKey []byte
var compositeEndKey []byte
compositeStartKey = historydb.ConstructPartialCompositeHistoryKey(namespace, key, false)
compositeEndKey = historydb.ConstructPartialCompositeHistoryKey(namespace, key, true)
// range scan to find any history records starting with namespace~key
dbItr := q.historyDB.db.GetIterator(compositeStartKey, compositeEndKey)
return newHistoryScanner(compositeStartKey, namespace, key, dbItr, q.blockStore), nil
}
可以看到，這里通過構(gòu)造compositeStartKey习蓬，compositeEndKey獲取指定范圍的iterator纽什。
瞄一下構(gòu)造key的方法
var compositeKeySep = []byte{0x00}
//ConstructPartialCompositeHistoryKey builds a partial History Key namespace~key~
// for use in history key range queries
func ConstructPartialCompositeHistoryKey(ns string, key string, endkey bool) []byte {
var compositeKey []byte
compositeKey = append(compositeKey, []byte(ns)...)
compositeKey = append(compositeKey, compositeKeySep...)
compositeKey = append(compositeKey, []byte(key)...)
compositeKey = append(compositeKey, compositeKeySep...)
if endkey {
compositeKey = append(compositeKey, []byte{0xff}...)
}
return compositeKey
}
注意endkey，前面說過躲叼，歷史數(shù)據(jù)是按(key=ns?key?blockNo?tranNo,value=nil)的格式存儲在leveldb上芦缰，這里的？指代的就是上面的分隔符[]byte{0x00}枫慷，endkey []byte{0xff}就是byte格式的最大值让蕾，這樣就能查詢出ns?key?開頭的所有key值。
繼續(xù)切入或听，最終定位到iterator的生成方法：
// GetIterator returns an iterator over key-value store. The iterator should be released after the use.
// The resultset contains all the keys that are present in the db between the startKey (inclusive) and the endKey (exclusive).
// A nil startKey represents the first available key and a nil endKey represent a logical key after the last available key
func (dbInst *DB) GetIterator(startKey []byte, endKey []byte) iterator.Iterator {
return dbInst.db.NewIterator(&goleveldbutil.Range{Start: startKey, Limit: endKey}, dbInst.readOpts)
}
注釋里有對iterator的startkey,endkey不同情況的詳細(xì)說明探孝。一個(gè)是對iterator區(qū)間是封前不封后，二是如果startKey為nil表示區(qū)間從第一個(gè)可用值開始誉裆，endKey為nil表示區(qū)間以最后一個(gè)有效值的后一位結(jié)束再姑，還有就是iterator不用的話要close()。
源碼改造
好了找御，函數(shù)調(diào)用過程和具體實(shí)現(xiàn)都已解析完畢元镀，接下來就是改源碼绍填，實(shí)現(xiàn)需求了。
實(shí)現(xiàn)分頁栖疑，無非就是拓展GetHistoryForKey()方法讨永，個(gè)人建議另外聲明一個(gè)函數(shù)實(shí)現(xiàn)而不對原函數(shù)做修改。
簡單實(shí)現(xiàn)遇革，直接新建接口GetHistoryForKeyByPage()卿闹，并在入?yún)⒅屑尤敕猪撔枰膮?shù)，如下：
GetHistoryForKeyByPage(key string, currentPage int64, pageSize int64) (HistoryQueryIteratorInterface, error)
這里僅僅加入當(dāng)前頁和頁容量兩個(gè)參數(shù)萝快，如果有別的需求可以直接改為傳入一個(gè)通用的option結(jié)構(gòu)體锻霎。
之后就是參照GetHistoryForKey()函數(shù)相繼增加后繼的實(shí)現(xiàn)函數(shù)。注意Client端調(diào)用的函數(shù)增加很簡單揪漩，直接仿照GetHistoryForKey()實(shí)現(xiàn)即可旋恼，Server端的修改涉及的內(nèi)容較多，一個(gè)是chaincode_shim.proto文件修改：
最終的實(shí)現(xiàn)放在leveldb_helper.go 冰更，具體實(shí)現(xiàn)筆者就不放上來啦，簡單思路就是遍歷iterator昂勒，根據(jù)currentPage蜀细，pageSize做截取。
編譯部署
最后一環(huán)節(jié)就是編譯部署戈盈，修改源碼后必須要重新編譯打包成docker奠衔，再次部署才能生效。
里面命令較多塘娶，有興趣可以都試試涣觉，涉及到重編譯和生成docker的已在圖中標(biāo)出，例如血柳，修改了peer工程下的代碼官册，編譯&docker生成只要 執(zhí)行：
make peer && make peer-docker
但是我們這里修改的文件大多在core目錄下，少量common目錄难捌，那就要執(zhí)行：
make clean && make docker
測試
兩種測試方法，一種是前面說的击敌，在historyleveldb_test.go中寫單元測試函數(shù)介返，可仿照TestHistory()對新分頁函數(shù)做測試，一種是編寫新chaincode，部署后通過cli或者sdk測試圣蝎，具體可參見我的上一篇博客 http://www.reibang.com/p/e16345cc2cde
tips:
追加需求刃宵，對歷史數(shù)據(jù)增加按時(shí)間戳查詢條件，如何實(shí)現(xiàn)徘公？
??????
十秒鐘過了牲证，有思路了么，思路其實(shí)不難关面，重新構(gòu)造key的格式坦袍，追加timestamp字段，定位到historyleveldb.go commit()方法等太，修改key的構(gòu)建方式：
...
// for each transaction, loop through the namespaces and writesets
// and add a history record for each write
for _, nsRWSet := range txRWSet.NsRwSets {
ns := nsRWSet.NameSpace
for _, kvWrite := range nsRWSet.KvRwSet.Writes {
writeKey := kvWrite.Key
//composite key for history records is in the form ns~key~blockNo~tranNo
//compositeHistoryKey := historydb.ConstructCompositeHistoryKey(ns, writeKey, blockNo, tranNo)
//composite key for history records is in the form ns~key~timestamp~blockNo~tranNo
compositeHistoryKey := historydb.ConstructCompositeHistoryKeyTimestamp(ns, writeKey, chdr.GetTimestamp(),blockNo, tranNo)
// No value is required, write an empty byte array (emptyValue) since Put() of nil is not allowed
dbBatch.Put(compositeHistoryKey, emptyValue)
}
}
...
當(dāng)然捂齐，之后的查詢實(shí)現(xiàn)都要做修改。
