handler_go.md

handler.go

主要提供：

• Handler 结构体：对所关联的链码容器进行相应，内部有一个状态机。
• HandleChaincodeStream() 方法：对外提供初始化的 Handler 结构体，并进入循环，不断接收来自链码容器的消息。

Handler 结构体

Peer 侧会为每一个 chaincode 维护一个 Handler 结构，具体响应所绑定的 chaincode 容器过来的各种消息，通过内部状态机进行处理。

Handler 结构实现了 MessageHandler 接口，主要提供一个 HandleMessage(msg *pb.ChaincodeMessage) error 方法，作为处理各个消息的入口方法。

type Handler struct {
	sync.RWMutex
	//peer to shim grpc serializer. User only in serialSend
	serialLock  sync.Mutex
	ChatStream  ccintf.ChaincodeStream
	FSM         *fsm.FSM
	ChaincodeID *pb.ChaincodeID
	ccInstance  *sysccprovider.ChaincodeInstance

	chaincodeSupport *ChaincodeSupport
	registered       bool
	readyNotify      chan bool
	// Map of tx txid to either invoke tx. Each tx will be
	// added prior to execute and remove when done execute
	txCtxs map[string]*transactionContext

	txidMap map[string]bool

	// used to do Send after making sure the state transition is complete
	nextState chan *nextStateInfo
}

chaincode 容器启动后，会调用到服务端的 Register() 方法，该方法进一步调用到 HandleChaincodeStream()，创建 Handler 结构体，进入接收消息循环。

// HandleChaincodeStream Main loop for handling the associated Chaincode stream
func HandleChaincodeStream(chaincodeSupport *ChaincodeSupport, ctxt context.Context, stream ccintf.ChaincodeStream) error {
    deadline, ok := ctxt.Deadline()
    chaincodeLogger.Debugf("Current context deadline = %s, ok = %v", deadline, ok)
    handler := newChaincodeSupportHandler(chaincodeSupport, stream)
    return handler.processStream()
}

newChaincodeSupportHandler 方法中会初始化 FSM。

之后，调用 handler.processStream() 进入对来自 chaincode 容器消息处理的主循环。

handler.processStream() 主消息循环

Peer 侧维护一个到 cc 的双向流，循环处理消息。主要在 func (handler *Handler) processStream() error 方法中（cc 到 peer 注册后会自动调用到该方法）。

主循环过程代码如下：

    for {
		in = nil
		err = nil
		nsInfo = nil
		if recv {
			recv = false
			go func() {
				var in2 *pb.ChaincodeMessage
				in2, err = handler.ChatStream.Recv()
				msgAvail <- in2
			}()
		}
		select {
		case sendErr := <-errc:
			if sendErr != nil {
				return sendErr
			}
			//send was successful, just continue
			continue
		case in = <-msgAvail:
			// Defer the deregistering of the this handler.
			if err == io.EOF {
				err = errors.Wrapf(err, "received EOF, ending chaincode support stream")
				chaincodeLogger.Debugf("%+v", err)
				return err
			} else if err != nil {
				chaincodeLogger.Errorf("Error handling chaincode support stream: %+v", err)
				return err
			} else if in == nil {
				err = errors.New("received nil message, ending chaincode support stream")
				chaincodeLogger.Debugf("%+v", err)
				return err
			}
			chaincodeLogger.Debugf("[%s]Received message %s from shim", shorttxid(in.Txid), in.Type.String())
			if in.Type.String() == pb.ChaincodeMessage_ERROR.String() {
				chaincodeLogger.Errorf("Got error: %s", string(in.Payload))
			}

			// we can spin off another Recv again
			recv = true

			if in.Type == pb.ChaincodeMessage_KEEPALIVE {
				chaincodeLogger.Debug("Received KEEPALIVE Response")
				// Received a keep alive message, we don't do anything with it for now
				// and it does not touch the state machine
				continue
			}
		case nsInfo = <-handler.nextState:
			in = nsInfo.msg
			if in == nil {
				err = errors.New("next state nil message, ending chaincode support stream")
				chaincodeLogger.Debugf("%+v", err)
				return err
			}
			chaincodeLogger.Debugf("[%s]Move state message %s", shorttxid(in.Txid), in.Type.String())
		case <-handler.waitForKeepaliveTimer():
			if handler.chaincodeSupport.keepalive <= 0 {
				chaincodeLogger.Errorf("Invalid select: keepalive not on (keepalive=%d)", handler.chaincodeSupport.keepalive)
				continue
			}

			//if no error message from serialSend, KEEPALIVE happy, and don't care about error
			//(maybe it'll work later)
			handler.serialSendAsync(&pb.ChaincodeMessage{Type: pb.ChaincodeMessage_KEEPALIVE}, nil)
			continue
		}

		err = handler.handleMessage(in)
		if err != nil {
			err = errors.WithMessage(err, "error handling message, ending stream")
			chaincodeLogger.Errorf("[%s] %+v", shorttxid(in.Txid), err)
			return err
		}

		if nsInfo != nil && nsInfo.sendToCC {
			chaincodeLogger.Debugf("[%s]sending state message %s", shorttxid(in.Txid), in.Type.String())
			//ready messages are sent sync
			if nsInfo.sendSync {
				if in.Type.String() != pb.ChaincodeMessage_READY.String() {
					panic(fmt.Sprintf("[%s]Sync send can only be for READY state %s\n", shorttxid(in.Txid), in.Type.String()))
				}
				if err = handler.serialSend(in); err != nil {
					return errors.WithMessage(err, fmt.Sprintf("[%s]error sending ready  message, ending stream:", shorttxid(in.Txid)))
				}
			} else {
				//if error bail in select
				handler.serialSendAsync(in, errc)
			}
		}
	}

首先是利用 select 结构尝试读取各种消息。包括：

• case in = <-msgAvail：从 cc 侧读取到请求消息；
• case nsInfo = <-handler.nextState：读取切换到下个状态的附加消息。
• case <-handler.waitForKeepaliveTimer()：定期发出心跳刷新消息。

读取到合法消息后，会分别调用 handler.HandleMessage(in) 处理 cc 消息；以及检查状态切换消息（仅允许消息类型为 READY，意味着此时 cc 在正常运行状态），是否要发送给 cc 侧（sendToCC 为 True）。

FSM

定义的状态、事件主要在 func newChaincodeSupportHandler(chaincodeSupport *ChaincodeSupport, peerChatStream ccintf.ChaincodeStream) *Handler 方法中。

一般对应 GET_STATE、GET_STATE_BY_RANGE 等简单事件，调用 handleXXX 方法。

PUT_STATE、DEL_STATE、INVOKE_CHAINCODE 三个事件，则会触发 enterBusyState() 方法。