Session类
我们终于到了关键地方，Session类代码位于libp2p\Session.h中 ，这个类代表的是socket通讯中的连接，这才是真正处理通讯协议的地方！
在上一节中Host::startPeerSession()函数将EthereumPeer类对象放入了Session类对象，具体在哪里呢？在这里：
1
| std::map<CapDesc, std::shared_ptr<Capability>> m_capabilities;
|
这个map中存放了这个连接的消息处理器。
我们再来看Session类是怎么工作的，首先从Session::start()看起。
1
2
3
4
5
| void Session::start()
{
ping();
doRead();
}
|
这个函数很简单，连接建立起来以后，先ping一下对方，也就是先给对方发一个ping的包，然后等待对方回应。
ping()函数没什么说的，我们来看doRead()这个函数：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
| void Session::doRead()
{
m_data.resize(h256::size);
ba::async_read(m_socket->ref(), boost::asio::buffer(m_data, h256::size),
[this, self](boost::system::error_code ec, std::size_t length) {
if (!checkRead(h256::size, ec, length))
return;
else if (!m_io->authAndDecryptHeader(bytesRef(m_data.data(), length)))
{
cnetlog << "header decrypt failed";
drop(BadProtocol);
return;
}
uint16_t hProtocolId;
uint32_t hLength;
uint8_t hPadding;
try
{
RLPXFrameInfo header(bytesConstRef(m_data.data(), length));
hProtocolId = header.protocolId;
hLength = header.length;
hPadding = header.padding;
}
catch (std::exception const& _e)
{
return;
}
auto tlen = hLength + hPadding + h128::size;
m_data.resize(tlen);
ba::async_read(m_socket->ref(), boost::asio::buffer(m_data, tlen),
[this, self, hLength, hProtocolId, tlen](
boost::system::error_code ec, std::size_t length) {
if (!checkRead(tlen, ec, length))
return;
else if (!m_io->authAndDecryptFrame(bytesRef(m_data.data(), tlen)))
{
cnetlog << "frame decrypt failed";
drop(BadProtocol);
return;
}
bytesConstRef frame(m_data.data(), hLength);
if (!checkPacket(frame))
{
return;
}
else
{
auto packetType = (PacketType)RLP(frame.cropped(0, 1)).toInt<unsigned>();
RLP r(frame.cropped(1));
bool ok = readPacket(hProtocolId, packetType, r);
if (!ok)
cnetlog << "Couldn't interpret packet. " << RLP(r);
}
doRead();
});
});
}
|
这段代码看起来比较长，可别被它吓到了，其实很简单，可以分成几步来看：
- 初始化接收缓冲区
1
| m_data.resize(h256::size);
|
- 收包
1
2
| ba::async_read(m_socket->ref(), boost::asio::buffer(m_data, h256::size),
[this, self](boost::system::error_code ec, std::size_t length)
|
- 收到包后检查包
1
2
3
4
5
6
7
8
| if (!checkRead(h256::size, ec, length))
return;
else if (!m_io->authAndDecryptHeader(bytesRef(m_data.data(), length)))
{
cnetlog << "header decrypt failed";
drop(BadProtocol);
return;
}
|
- 没问题的话解析包头
1
2
3
4
| RLPXFrameInfo header(bytesConstRef(m_data.data(), length));
hProtocolId = header.protocolId;
hLength = header.length;
hPadding = header.padding;
|
hLength表示包头后续数据包大小，hPadding表示后续数据包与包头间的偏移。
- 接收后续数据包
1
2
3
4
| auto tlen = hLength + hPadding + h128::size;
m_data.resize(tlen);
ba::async_read(m_socket->ref(), boost::asio::buffer(m_data, tlen),
[this, self, hLength, hProtocolId, tlen]
|
可以看到后续数据包大小的计算方法。
- 再次检查数据包
这个和第三步是一样的
- 解析数据包
1
2
3
4
| bytesConstRef frame(m_data.data(), hLength);
auto packetType = (PacketType)RLP(frame.cropped(0, 1)).toInt<unsigned>();
RLP r(frame.cropped(1));
bool ok = readPacket(hProtocolId, packetType, r);
|
数据包中的内容放到了RLP类对象中，然后调用readPacket()进行具体分析。
- 继续收包
Session::readPacket()
这里就是具体处理网络数据包的地方了，代码也不多：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
| bool Session::readPacket(uint16_t _capId, PacketType _t, RLP const& _r)
{
try
{
if (_capId == 0 && _t < UserPacket)
return interpret(_t, _r);
for (auto const& i: m_capabilities)
if (_t >= (int)i.second->m_idOffset && _t - i.second->m_idOffset < i.second->hostCapability()->messageCount())
return i.second->enabled() ? i.second->interpret(_t - i.second->m_idOffset, _r) : true;
return false;
}
catch (std::exception const& _e)
{
return true;
}
return true;
}
|
其中的_t类型是PacketType，其实就是上一节所说的消息偏移，消息偏移小于UserPacket的包由Session类自身处理，剩下的包交给m_capabilities中的消息处理器，每个消息处理器处理各自能力范围内的消息。
最后看下Session::interpret()函数的实现：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
| bool Session::interpret(PacketType _t, RLP const& _r)
{
switch (_t)
{
case DisconnectPacket:
{
break;
}
case PingPacket:
{
cnetdetails << "Ping " << m_info.id;
RLPStream s;
sealAndSend(prep(s, PongPacket));
break;
}
case PongPacket:
DEV_GUARDED(x_info)
{
m_info.lastPing = std::chrono::steady_clock::now() - m_ping;
cnetdetails << "Latency: "
<< chrono::duration_cast<chrono::milliseconds>(m_info.lastPing).count()
<< " ms";
}
break;
}
return true;
}
|
可以看到这里真正有用的地方只是收到ping包之后回复一个pong包，和前面的握手流程类似。收到pong包以后也只是更新了时间，没有其他内容了，那第一个真正有意义的包是谁来发起的呢？这个需要从Session类外去寻找答案了。
