解读gobgp（三）

func (h *fsmHandler) established(ctx context.Context) (bgp.FSMState, *fsmStateReason) {
    ...

	go h.sendMessageloop(ctx, &wg)
	h.msgCh = h.incoming // 这个incoming来自于fsm的incoming
    // 这里会循环接收conn来的BGP消息，这里面会调用recvMessageWithError
	go h.recvMessageloop(ctx, &wg)
    ...

func (h *fsmHandler) recvMessageWithError() (*fsmMsg, error) {
	...
    // 这里反序列化BGP消息
	headerBuf, err := readAll(h.conn, bgp.BGP_HEADER_LENGTH)
	if err != nil {
		sendToStateReasonCh(fsmReadFailed, nil)
		return nil, err
	}

	hd := &bgp.BGPHeader{}
	err = hd.DecodeFromBytes(headerBuf)
	...

	bodyBuf, err := readAll(h.conn, int(hd.Len)-bgp.BGP_HEADER_LENGTH)
	...

	switch handling {
	case bgp.ERROR_HANDLING_AFISAFI_DISABLE:
		fmsg.MsgData = m
		return fmsg, nil
	case bgp.ERROR_HANDLING_SESSION_RESET:
		...
		fmsg.MsgData = err
		return fmsg, err
	default:
		...

		if establishedState {
			switch m.Header.Type {
			case bgp.BGP_MSG_ROUTE_REFRESH:
				fmsg.MsgType = fsmMsgRouteRefresh
			case bgp.BGP_MSG_UPDATE:
				...
				fallthrough
			case bgp.BGP_MSG_KEEPALIVE:
				// 注意ka消息不会被server处理
				if m.Header.Type == bgp.BGP_MSG_KEEPALIVE {
					return nil, nil
				}
			case bgp.BGP_MSG_NOTIFICATION:
				...
				return nil, nil
			}
		}
	}
    // 返回的fmsg会被送入fsmHandler的msgCh中
	return fmsg, nil
}

func (s *BgpServer) Serve() {
    ...

	for {
        // 这里会扫描api接口消息和各个peer fsm的消息
		cases := make([]reflect.SelectCase, firstPeerCaseIndex+len(s.incomings))
		...

		chosen, value, ok := reflect.Select(cases)
		switch chosen {
		case 0: // 处理api消息
			op := value.Interface().(*mgmtOp)
			s.handleMGMTOp(op)
		case 1: // 处理连接
			conn := value.Interface().(*net.TCPConn)
			s.passConnToPeer(conn)
		case 2: // 处理ROA事件
			ev := value.Interface().(*roaEvent)
			s.roaManager.HandleROAEvent(ev)
		default: // 这里会处理BGP消息
			if ok {
				e := value.Interface().(*fsmMsg)
				handlefsmMsg(e)
			}
		}
	}
}