100 lines
2.3 KiB
Go
100 lines
2.3 KiB
Go
// Package protocol 实现MongoDB协议解析
|
|
package protocol
|
|
|
|
import (
|
|
"encoding/binary"
|
|
"fmt"
|
|
)
|
|
|
|
// Message MongoDB协议消息结构
|
|
type Message struct {
|
|
Header Header
|
|
OpCode OpCode
|
|
OriginalBody []byte // 原始消息体（解析前）
|
|
Body interface{} // 解析后的消息体
|
|
}
|
|
|
|
// Header 消息头
|
|
type Header struct {
|
|
MessageLength int32
|
|
RequestID int32
|
|
ResponseTo int32
|
|
OpCode OpCode
|
|
}
|
|
|
|
// OpCode 操作码
|
|
type OpCode int32
|
|
|
|
const (
|
|
OP_REPLY OpCode = 1
|
|
OP_MSG OpCode = 2
|
|
OP_UPDATE OpCode = 2001
|
|
OP_INSERT OpCode = 2002
|
|
RESERVED OpCode = 2003
|
|
OP_QUERY OpCode = 2004
|
|
OP_GET_MORE OpCode = 2005
|
|
OP_DELETE OpCode = 2006
|
|
OP_KILL_CURSORS OpCode = 2007
|
|
OP_COMMAND OpCode = 2010
|
|
OP_COMMAND_REPLY OpCode = 2011
|
|
OP_COMPRESSED OpCode = 2012
|
|
OP_ENCRYPTED OpCode = 2013
|
|
)
|
|
|
|
// ParseMessage 解析MongoDB协议消息
|
|
func ParseMessage(data []byte) (*Message, error) {
|
|
if len(data) < 16 {
|
|
return nil, fmt.Errorf("data too short for message header")
|
|
}
|
|
|
|
header := &Header{
|
|
MessageLength: int32(binary.LittleEndian.Uint32(data[0:4])),
|
|
RequestID: int32(binary.LittleEndian.Uint32(data[4:8])),
|
|
ResponseTo: int32(binary.LittleEndian.Uint32(data[8:12])),
|
|
OpCode: OpCode(binary.LittleEndian.Uint32(data[12:16])),
|
|
}
|
|
|
|
body := data[16:]
|
|
|
|
// 解析特定操作码的消息体
|
|
var parsedBody interface{}
|
|
switch header.OpCode {
|
|
case OP_QUERY:
|
|
query, err := parseQuery(body)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
parsedBody = query
|
|
case OP_INSERT:
|
|
insert, err := parseInsert(body)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
parsedBody = insert
|
|
// 这里可以添加更多操作码的解析逻辑
|
|
default:
|
|
// 未知操作码，保留原始数据
|
|
parsedBody = body
|
|
}
|
|
|
|
return &Message{
|
|
Header: *header,
|
|
OpCode: header.OpCode,
|
|
OriginalBody: body,
|
|
Body: parsedBody,
|
|
}, nil
|
|
}
|
|
|
|
// 解析查询请求
|
|
func parseQuery(data []byte) (interface{}, error) {
|
|
// 实现具体的查询消息解析逻辑
|
|
// 这里返回原始数据作为占位符
|
|
return data, nil
|
|
}
|
|
|
|
// 解析插入请求
|
|
func parseInsert(data []byte) (interface{}, error) {
|
|
// 实现具体的插入消息解析逻辑
|
|
// 这里返回原始数据作为占位符
|
|
return data, nil
|
|
} |