深入理解Node.js的Inspector

前言：Node.js提供的Inspector不仅可以用来调试Node.js代码，还可以实时收集Node.js进程的内存，CPU等数据，同时支持静态、动态开启，是一个非常强大的工具，本文从使用和原理详细讲解Inspector。

Node.js的文档中对inspector的描述很少，但是如果深入探索，其实里面的内容还是挺多的。我们先看一下Inspector的使用。

1 Inspector的使用

1.1 本地调试

我们先从一个例子开始。下面是一个http服务器。

const http = require('http');
http.createServer((req, res) => {
res.end('ok');
}).listen(80);

然后我们以node --inspect httpServer.js的方式启动。我们可以看到以下输出。

Debugger listening on ws://127.0.0.1:9229/fbbd9d8f-e088-48cc-b1e0-e16bfe58db44
For help, see: https://nodejs.org/en/docs/inspector

9229端口是Node.js默认选择的端口，当然我们也可以自定义，具体可参考文档。这时候我们去浏览器打开开发者工具，菜单栏多了一个调试Node.js的按钮。

点击这个按钮。我们可以看到以下界面。

我们可以选择某一行代码打断点，比如我在第三行，这时候我们访问80端口，开发者工具就会停留在断点处。这时候我们可以看到一些执行上下文。

1.2 远程调试

但很多时候我们可能需要远程调试。比如我在一台云服务器上部署以上服务器代码。然后执行

node --inspect=0.0.0.0:8888 httpServer.js

不过这时候我们打开开发者工具就会发现按钮置灰或者找不到我们远程服务器的信息。这时候我们需要用另一种方式。通过在浏览器url输入框输入devtools://devtools/bundled/js_app.html?experiments=true&v8only=true&ws={host}:{port}/{path}的方式(替换{}里面的内容为你执行Node.js时输出的信息)，浏览器就会去连接你输入的地址，比如1.1.1.1:9229/abc。这种比较适合于对于通用的场景。

1.3 自动探测

如果是我们自己调试的话，这种方式看起来就有点麻烦，我们可以使用浏览器提供的自动探测功能。1 url输入框输入chrome://inspect/#devices我们会看到以下界面

2 点击configure按钮，在弹出的弹框里输入你远程服务器的地址

3 配置完毕后，我们会看到界面变成这样了，或者打开新的tab，我们看到开发者工具的调试按钮也变亮了。

4 这时候我们点击inspect按钮、Open dedicated DevTools for Node按钮或者打开新tab的开发者工具，就可以开始调试。而且还可以调试Node.js的原生js模块。

2 Inspector调试的原理

下面以通过url的方式调试(可以看到network)，来看看调试的时候都发生了什么，浏览器和远程服务器建立连接后，是通过websocket协议通信的。

我们看一下这命令是什么意思，首先看Debugger.scriptParsed。

Debugger.scriptParsed # Fired when virtual machine parses script. This event is also fired for all known and uncollected scripts upon enabling debugger.

从说明中我们看到，当V8解析脚本的时候就会触发这个事件，那就会告诉浏览器这个信息。

我们发现返回的都是一些元数据，没有脚本的具体代码内容，这时候浏览器会再次发起请求，

我们看到这个脚本的scriptId是103。所以请求里带了这个scriptId。对应的请求id是11。接着看一下响应。

至此，我们了解了获取脚本内容的过程，然后我们看看调试的时候是怎样的过程。当我们在浏览器上点击某一行设置断点的时候，浏览器就会发送一个请求。

这个命令的意义顾名思义，我们看一下具体定义。

Debugger.setBreakpointByUrl # Sets JavaScript breakpoint at given location specified either by URL or URL regex. Once this command is issued, all existing parsed scripts will have breakpoints resolved and returned in locations property. Further matching script parsing will result in subsequent breakpointResolved events issued. This logical breakpoint will survive page reloads.

接着服务返回响应。

这时候我们从另外一个tab访问80端口。服务器就会在我们设置的断点处停留，并且通知浏览器。

我们看一下这个命令的意思。

这个命令就是当服务器执行到断点时通知浏览器，并且返回执行的一些上下文，比如是哪个执行到哪个断点停留了。这时候浏览器侧也会停留在对应的地方，当我们hover某个变量时，就会看到对应的上下文。这些都是通过具体的命令获取的数据。就不一一分析了，可以参考具体文档。

3 Inspector的实现

大致了解了浏览器和服务器的交互过程和协议后，我们再来深入了解一下关于inspector的一些实现。当然这里不是分析V8中Inspector的实现，而是分析如何使用V8的Inspector以及Node.js中关于Inspector的实现部分。

3.1 开源实现

因为Node.js的实现比较复杂，这里先以一个简单版的调试工具源码来分析inspector的原理。我们先看一下初始化代码。

inspector = std::unique_ptr<Inspector>(new Inspector(v8Platform, context, port));
inspector->startAgent();

首先新建一个Inspector。然后启动它。接下来看看Inspector里的逻辑。

Inspector::Inspector(
const std::unique_ptr<v8::Platform> &platform,
const v8::Local<v8::Context> &context,
const int webSocketPort) {
context_ = context;
// 新建一个websocket server用于和客户端通信
websocket_server_ = std::unique_ptr<WebSocketServer>(
new WebSocketServer(
webSocketPort,
// 收到客户的的消息后执行onMessage回调
std::bind(&Inspector::onMessage, this, std::placeholders::_1)
)
);
// 新建一个inspector client和V8通信
inspector_client_ = std::unique_ptr<V8InspectorClientImpl>(
new V8InspectorClientImpl(
platform,
context_,
// 收到V8的消息后调用sendMessage回复给客户的
std::bind(&Inspector::sendMessage, this, std::placeholders::_1),
std::bind(&Inspector::waitForFrontendMessage, this)
)
);
}

代码看起来很复杂，不过我们不需要深究。主要是两个部分，一个是新建一个websocket服务器，一个是新建一个inspector客户端(用于和V8 Inspector通信)，整体架构如下。

接下来分别看一下websocket服务器和inspector客户端的实现。首先看一下websocket服务器的构造函数。

WebSocketServer::WebSocketServer(int port, std::function<void(std::string)> onMessage){
port_ = port;
onMessage_ = std::move(onMessage);
}

WebSocketServer构造函数的实现很简单，只是初始化一些字段。接着看inspector客户端的实现。

V8InspectorClientImpl:: V8InspectorClientImpl(const std::unique_ptr<v8::Platform> &platform, const v8::Local<v8::Context> &context, const std::function<void(std::string)> &onResponse, const std::function<int(void)> &onWaitFrontendMessageOnPause) {
platform_ = platform.get();
context_ = context;
onWaitFrontendMessageOnPause_ = onWaitFrontendMessageOnPause;
isolate_ = context_->GetIsolate();
// 创建一个channel和inspector通信，收到V8消息时会执行onResponse
channel_.reset(new V8InspectorChannelImp(isolate_, onResponse));
// 新建一个V8提供的inspector
inspector_ = v8_inspector::V8Inspector::create(isolate_, this);
// 创建一个和inspector通信的session。
session_ = inspector_->connect(kContextGroupId, channel_.get(), v8_inspector::StringView());
context_->SetAlignedPointerInEmbedderData(1, this);
v8_inspector::StringView contextName = convertToStringView("inspector");
inspector_->contextCreated(v8_inspector::V8ContextInfo(context, kContextGroupId, contextName));
terminated_ = true;
run_nested_loop_ = false;
}

上面代码很多，主要是根据V8提供的API来就行。这里主要有三个概念

1 V8Inspector是V8提供的类。

2 session表示和V8 inspector通信的会话。

3 channel用于和V8 inspector通信，从API来看，channel只能从V8获取数据，写入数据是另外的API。

这时候的架构如下

至此，websocket服务器和inspector客户端就分析完毕了，回到最开始的代码，初始化完毕后会执行startAgent。

void Inspector::startAgent() {
websocket_server_->run();
}
// 启动websocket服务器
void WebSocketServer::run() {
auto const address = net::ip::make_address("127.0.0.1");
net::io_context ioc{1};
tcp::acceptor acceptor{ioc, {address, static_cast<unsigned short>(port_)}};
tcp::socket socket{ioc};
acceptor.accept(socket);
ws_ = std::unique_ptr<websocket::stream<tcp::socket>>(new websocket::stream<tcp::socket>(std::move(socket)));
startListening();
}
// 等待连接
void WebSocketServer::startListening(){
ws_->accept();
while (true) {
waitFrontendMessage();
}}// 读取连接中的消息void WebSocketServer::waitFrontendMessage(){
beast::flat_buffer buffer;
ws_->read(buffer);
std::string message = boost::beast::buffers_to_string(buffer.data());
onMessage_(std::move(message));
}

startAgent的逻辑就是启动websocket服务器。启动完毕后就等待客户的连接。连接成功后执行onMessage_。我们看一下onMessage的实现。

void Inspector::onMessage(const std::string& message) {
std::cout << "CDT message: " << message << std::endl;
// StringView是V8要求的格式
v8_inspector::StringView protocolMessage = convertToStringView(message);
// 通知V8 Inspector
inspector_client_->dispatchProtocolMessage(protocolMessage);
}

onMessage通过Inspector客户端把消息交给V8 Inspector处理。V8 Inspector处理完后，通过channel通知Inspector客户端，对应的函数是sendResponse。V8InspectorChannelImp是继承V8提供的Channel，sendResponse是一个纯虚函数，由V8InspectorChannelImp实现。

void V8InspectorChannelImp::sendResponse(int callId, std::unique_ptr<v8_inspector::StringBuffer> message) {
const std::string response = convertToString(isolate_, message->string());
onResponse_(response);
}

onResponse_是在Chnnel初始化时设置的，对应函数是inspector客户端的sendMessage。

void Inspector::sendMessage(const std::string& message) {
websocket_server_->sendMessage(message);
}

sendMessage通过websocket服务器把V8 Inspector返回的消息返回给客户的。至此，整个通信流程就完成了。

3.2 Node.js的实现(v14)

Node.js的实现非常复杂并且很绕，本文根据也无法通俗易懂地介绍和分析，只能按照我自己的思路大致讲解一下流程，有兴趣的同学可以自行阅读圆源码。

3.2.1 初始化

在Node.js初始化的时候，会创建一个inspector::Agent并启动。

inspector_agent_ = std::make_unique<inspector::Agent>(this);
env->InitializeInspector({});

接着看InitializeInspector。

inspector_agent_->Start(inspector_path,
options_->debug_options(),
inspector_host_port(),
is_main_thread());

Start中会执行StartIoThread进行进一步操作。

bool Agent::StartIoThread() {
io_ = InspectorIo::Start(client_->getThreadHandle(),
path_,
host_port_,
debug_options_.inspect_publish_uid);
return true;
}

继续看InspectorIo::Start

std::unique_ptr<InspectorIo> InspectorIo::Start(
std::shared_ptr<MainThreadHandle> main_thread,
const std::string& path,
std::shared_ptr<HostPort> host_port,
const InspectPublishUid& inspect_publish_uid) {
auto io = std::unique_ptr<InspectorIo>(
new InspectorIo(main_thread,
path,
host_port,
inspect_publish_uid));
return io;
}

InspectorIo::Start其实是创建了一个InspectorIo对象，我们看一下构造函数里的逻辑。

InspectorIo::InspectorIo(std::shared_ptr<MainThreadHandle> main_thread,
const std::string& path,
std::shared_ptr<HostPort> host_port,
const InspectPublishUid& inspect_publish_uid)
: main_thread_(main_thread),
host_port_(host_port),
inspect_publish_uid_(inspect_publish_uid),
thread_(),
script_name_(path),
id_(GenerateID()) {
Mutex::ScopedLock scoped_lock(thread_start_lock_);
CHECK_EQ(uv_thread_create(&thread_, InspectorIo::ThreadMain, this), 0);
thread_start_condition_.Wait(scoped_lock);
}

构造函数里创建了一个线程并在线程中执行InspectorIo::ThreadMain。我们接着看ThreadMain。

void InspectorIo::ThreadMain(void* io) {
static_cast<InspectorIo*>(io)->ThreadMain();
}
void InspectorIo::ThreadMain() {
uv_loop_t loop;
loop.data = nullptr;
int err = uv_loop_init(&loop);
CHECK_EQ(err, 0);
std::shared_ptr<RequestQueueData> queue(new RequestQueueData(&loop),
RequestQueueData::CloseAndFree);
std::string script_path = ScriptPath(&loop, script_name_);
// 创建一个处理请求的delegate对象
std::unique_ptr<InspectorIoDelegate> delegate(
new InspectorIoDelegate(queue, main_thread_, id_,
script_path, script_name_));
// 创建一个服务器
InspectorSocketServer server(std::move(delegate),
&loop,
host_port_->host(),
host_port_->port(),
inspect_publish_uid_);
request_queue_ = queue->handle();
// Its lifetime is now that of the server delegate
queue.reset();
{
Mutex::ScopedLock scoped_lock(thread_start_lock_);
// 启动服务器
if (server.Start()) {
host_port_->set_port(server.Port());
}
thread_start_condition_.Broadcast(scoped_lock);
}
// 进入事件循环
uv_run(&loop, UV_RUN_DEFAULT);
CheckedUvLoopClose(&loop);
}

我们分析一下服务器相关的逻辑。先看一下创建一个服务器的逻辑

InspectorSocketServer::InspectorSocketServer(
std::unique_ptr<SocketServerDelegate> delegate, uv_loop_t* loop,
const std::string& host, int port,
const InspectPublishUid& inspect_publish_uid, FILE* out)
: loop_(loop),
delegate_(std::move(delegate)),
host_(host),
port_(port),
inspect_publish_uid_(inspect_publish_uid),
next_session_id_(0),
out_(out) {
// 把服务器对象关联到delete对象中
delegate_->AssignServer(this);
// 状态为初始化
state_ = ServerState::kNew;
}

接着看启动服务器的逻辑。

bool InspectorSocketServer::Start() {
std::unique_ptr<SocketServerDelegate> delegate_holder;
// We will return it if startup is successful
delegate_.swap(delegate_holder);
struct addrinfo hints;
memset(&hints, 0, sizeof(hints));
hints.ai_flags = AI_NUMERICSERV;
hints.ai_socktype = SOCK_STREAM;
uv_getaddrinfo_t req;
const std::string port_string = std::to_string(port_);
// 获取地址信息
int err = uv_getaddrinfo(loop_, &req, nullptr, host_.c_str(),
port_string.c_str(), &hints);
for (addrinfo* address = req.addrinfo; address != nullptr;
address = address->ai_next) {
// 真正创建一个服务器
auto server_socket = ServerSocketPtr(new ServerSocket(this));
// 监听地址，服务器启动
err = server_socket->Listen(address->ai_addr, loop_);
if (err == 0)
server_sockets_.push_back(std::move(server_socket));
}
uv_freeaddrinfo(req.addrinfo);
delegate_.swap(delegate_holder);
state_ = ServerState::kRunning;
// 打印提示，让用户知道连接到哪个url
PrintDebuggerReadyMessage(host_,
server_sockets_,
delegate_->GetTargetIds(),
inspect_publish_uid_.console,
out_);
return true;
}

启动服务器的本质就是监听某些IP+端口，所以我们接着分析一下ServerSocket对象的Listen函数。ServerSocket是对Libuv结构体uv_tcp_t的封装，表示一个基于TCP的socket。另外ServerSocket还保存了所属的InspectorSocketServer对象。

int ServerSocket::Listen(sockaddr* addr, uv_loop_t* loop) {
uv_tcp_t* server = &tcp_socket_;
// 初始化Libuv handle
CHECK_EQ(0, uv_tcp_init(loop, server));
// 在handle上绑定地址
uv_tcp_bind(server, addr, 0);
// 监听socket，监听成功后执行SocketConnectedCallback
uv_listen(reinterpret_cast<uv_stream_t*>(server), 511, ServerSocket::SocketConnectedCallback);
return err;
}

继续看SocketConnectedCallback

void ServerSocket::SocketConnectedCallback(uv_stream_t* tcp_socket,
int status) {
// 监听成功
if (status == 0) {
// uv_tcp_t获取所属的ServerSocket对象
ServerSocket* server_socket = ServerSocket::FromTcpSocket(tcp_socket);
// Memory is freed when the socket closes.
server_socket->server_->Accept(server_socket->port_, tcp_socket);
}
}

接着看InspectorSocketServer::Accept。

void InspectorSocketServer::Accept(int server_port,
uv_stream_t* server_socket) {
// 新建一个session表示一个连接对应的会话
std::unique_ptr<SocketSession> session(
new SocketSession(this, next_session_id_++, server_port));
// 申请一个Delegate处理连接中的数据
InspectorSocket::DelegatePointer delegate =
InspectorSocket::DelegatePointer(
new SocketSession::Delegate(this, session->id()));
// 摘取一个连接
InspectorSocket::Pointer inspector =
InspectorSocket::Accept(server_socket, std::move(delegate));
if (inspector) {
// 设置session对应的InspectorSocket
session->Own(std::move(inspector));
connected_sessions_[session->id()].second = std::move(session);
}
}

接着看InspectorSocket::Accept

InspectorSocket::Pointer InspectorSocket::Accept(uv_stream_t* server,
DelegatePointer delegate) {
// 摘取一个TCP连接
auto tcp = TcpHolder::Accept(server, std::move(delegate));
// 操作成功
if (tcp) {
// 新建一个InspectorSocket表示和客户的通信的对象
InspectorSocket* inspector = new InspectorSocket();
// 设置处理连接数据的协议handler，即把数据当作HTTP协议处理
inspector->SwitchProtocol(new HttpHandler(inspector, std::move(tcp)));
return InspectorSocket::Pointer(inspector);
} else {
return InspectorSocket::Pointer(nullptr);
}
}

接着看TcpHolder::Accept

TcpHolder::Pointer TcpHolder::Accept(
uv_stream_t* server,
InspectorSocket::DelegatePointer delegate) {
// 新建一个TcpHolder表示TCP连接
TcpHolder* result = new TcpHolder(std::move(delegate));
uv_stream_t* tcp = reinterpret_cast<uv_stream_t*>(&result->tcp_);
// 初始化
uv_tcp_init(server->loop, &result->tcp_);
// 这才是真正的摘取连接逻辑，uv_accept会把连接的fd保存到tcp结构体中
uv_accept(server, tcp);
// 设置等待可读事件，回调是OnDataReceivedCb
uv_read_start(tcp, allocate_buffer, OnDataReceivedCb);
}

当数据到来时，我们看一下OnDataReceivedCb是怎么处理的。

void TcpHolder::OnDataReceivedCb(uv_stream_t* tcp, ssize_t nread,
const uv_buf_t* buf) {
TcpHolder* holder = From(tcp);
holder->ReclaimUvBuf(buf, nread);
if (nread < 0 || nread == UV_EOF) {
holder->handler_->OnEof();
} else {
holder->handler_->OnData(&holder->buffer);
}
}

OnDataReceivedCb会调用handler_的onData函数，handler_就是HttpHandler。

void OnData(std::vector<char>* data) override {
llhttp_errno_t err;
// 解析http协议
err = llhttp_execute(&parser_, data->data(), data->size());
if (err == HPE_PAUSED_UPGRADE) {
err = HPE_OK;
llhttp_resume_after_upgrade(&parser_);
}
// 省略一些步骤，第一个请求是一个升级http协议到websocket协议的请求
delegate()->OnSocketUpgrade(event.host, event.path, event.ws_key);
}

看看HttpHandler的delegate(ProtocolHandler是HttpHandler的基类)

InspectorSocket::Delegate* ProtocolHandler::delegate() {
return tcp_->delegate();
}
InspectorSocket::Delegate* TcpHolder::delegate() {
return delegate_.get();
}

获取的是TcpHolder的delegate。而TcpHolder的delegate是SocketSession::Delegate。即最后调用的是SocketSession::Delegate的OnSocketUpgrade函数。

void SocketSession::Delegate::OnSocketUpgrade(const std::string& host,
const std::string& path,
const std::string& ws_key) {
std::string id = path.empty() ? path : path.substr(1);
server_->SessionStarted(session_id_, id, ws_key);
}

继续看server_->SessionStarted。

void InspectorSocketServer::SessionStarted(int session_id,
const std::string& id,
const std::string& ws_key) {
SocketSession* session = Session(session_id);
connected_sessions_[session_id].first = id;
// 处理wesocket和客户端的会话
session->Accept(ws_key);
// 开启一个服务器中agent和V8 inspector会话
delegate_->StartSession(session_id, id);
}

server_->SessionStarted主要有三个逻辑

1 保存一个和客户端通信会话的关系

2 完成http协议到websocket协议的升级

3 新建一个和V8 Inspector的会话，Node.js类似一个代理。

我们先看如何处理websocket和客户端的会话

void Accept(const std::string& ws_key) {
ws_socket_->AcceptUpgrade(ws_key);
}

ws_socket_是session对应的InspectorSocket对象。

void InspectorSocket::AcceptUpgrade(const std::string& ws_key) {
protocol_handler_->AcceptUpgrade(ws_key);
}

InspectorSocket::AcceptUpgrade根据当前处理协议的handler进一步处理，目前是HTTP协议。

void AcceptUpgrade(const std::string& accept_key) override {
char accept_string[ACCEPT_KEY_LENGTH];
generate_accept_string(accept_key, &accept_string);
// 回复101 Switching Protocols给客户端说明同意协议升级
const char accept_ws_prefix[] = "HTTP/1.1 101 Switching Protocols\r\n"
"Upgrade: websocket\r\n"
"Connection: Upgrade\r\n"
"Sec-WebSocket-Accept: ";
const char accept_ws_suffix[] = "\r\n\r\n";
std::vector<char> reply(accept_ws_prefix,
accept_ws_prefix + sizeof(accept_ws_prefix) - 1);
reply.insert(reply.end(), accept_string,
accept_string + sizeof(accept_string));
reply.insert(reply.end(), accept_ws_suffix,
accept_ws_suffix + sizeof(accept_ws_suffix) - 1);
if (WriteRaw(reply, WriteRequest::Cleanup) >= 0) {
// 切换协议为websocket，协议升级成功后，后续的数据被当作websocket协议处理
inspector_->SwitchProtocol(new WsHandler(inspector_, std::move(tcp_)));
} else {
tcp_.reset();
}
}

AcceptUpgrade完成了协议升级，接着服务器还需要和V8 inspector建立一个会话。

void InspectorIoDelegate::StartSession(int session_id,
const std::string& target_id) {
auto session = main_thread_->Connect(
std::unique_ptr<InspectorSessionDelegate>(
new IoSessionDelegate(request_queue_->handle(), session_id)), true);
if (session) {
sessions_[session_id] = std::move(session);
fprintf(stderr, "Debugger attached.\n");
}
}

main_thread_是MainThreadHandle对象。

std::unique_ptr<InspectorSession> MainThreadHandle::Connect(
std::unique_ptr<InspectorSessionDelegate> delegate,
bool prevent_shutdown) {
return std::unique_ptr<InspectorSession>(
new CrossThreadInspectorSession(++next_session_id_,
shared_from_this(),
std::move(delegate),
prevent_shutdown));
}

这里新建了一个CrossThreadInspectorSession对象。我们看看CrossThreadInspectorSessionde 构造函数。

CrossThreadInspectorSession(
int id,
std::shared_ptr<MainThreadHandle> thread,
std::unique_ptr<InspectorSessionDelegate> delegate,
bool prevent_shutdown)
: state_(thread, std::bind(MainThreadSessionState::Create,
std::placeholders::_1,
prevent_shutdown)) {
state_.Call(&MainThreadSessionState::Connect, std::move(delegate));
}

构造函数中执行了MainThreadSessionState::Connect函数

void Connect(std::unique_ptr<InspectorSessionDelegate> delegate) {
Agent* agent = thread_->inspector_agent();
if (agent != nullptr)
session_ = agent->Connect(std::move(delegate), prevent_shutdown_);
}

Connect最后调用了agent的Connect。

std::unique_ptr<InspectorSession> Agent::Connect(
std::unique_ptr<InspectorSessionDelegate> delegate,
bool prevent_shutdown) {
CHECK_NOT_NULL(client_);
int session_id = client_->connectFrontend(std::move(delegate),
prevent_shutdown);
return std::unique_ptr<InspectorSession>(
new SameThreadInspectorSession(session_id, client_));
}

继续看client_->connectFrontend

int connectFrontend(std::unique_ptr<InspectorSessionDelegate> delegate,
bool prevent_shutdown) {
int session_id = next_session_id_++;
channels_[session_id] = std::make_unique<ChannelImpl>(env_,
client_,
getWorkerManager(),
std::move(delegate),
getThreadHandle(),
prevent_shutdown);
return session_id;
}

新建了一个ChannelImpl对象，ChannelImpl维护了一个和V8 Inspector的会话。

explicit ChannelImpl(Environment* env,
const std::unique_ptr<V8Inspector>& inspector,
std::shared_ptr<WorkerManager> worker_manager,
std::unique_ptr<InspectorSessionDelegate> delegate,
std::shared_ptr<MainThreadHandle> main_thread_,
bool prevent_shutdown)
: delegate_(std::move(delegate)), prevent_shutdown_(prevent_shutdown),
retaining_context_(false) {
/// 连接到V8 inspector获得一个session
session_ = inspector->connect(CONTEXT_GROUP_ID, this, StringView());
}

至此就完成了连接建立，协议升级的处理。架构如下

3.2.2 客户端数据到V8 Inspector的处理流程

我们继续看业务数据的处理过程。我们刚才已经分析过，当连接上有数据到来时会执行OnDataReceivedCb。

void TcpHolder::OnDataReceivedCb(uv_stream_t* tcp, ssize_t nread,
const uv_buf_t* buf) {
TcpHolder* holder = From(tcp);
holder->ReclaimUvBuf(buf, nread);
if (nread < 0 || nread == UV_EOF) {
holder->handler_->OnEof();
} else {
holder->handler_->OnData(&holder->buffer);
}
}

这时候handler变成了websocket handler。

void OnData(std::vector<char>* data) override {
// 1. Parse.
int processed = 0;
do {
// 解析数据
processed = ParseWsFrames(*data);
// 2. Fix the data size & length
if (processed > 0) {
remove_from_beginning(data, processed);
}
} while (processed > 0 && !data->empty());
}

我们看一下解析数据的逻辑。

int ParseWsFrames(const std::vector<char>& buffer) {
int bytes_consumed = 0;
std::vector<char> output;
bool compressed = false;
// 按照websocket协议解析数据
ws_decode_result r = decode_frame_hybi17(buffer,
true /* client_frame */,
&bytes_consumed, &output,
&compressed);
delegate()->OnWsFrame(output);
}

delegate我们刚才已经分析过是SocketSession::Delegate。

void SocketSession::Delegate::OnWsFrame(const std::vector<char>& data) {
server_->MessageReceived(session_id_, std::string(data.data(), data.size()));
}

和处理协议升级时类似的逻辑。我们看server_->MessageReceived。

void MessageReceived(int session_id, const std::string& message) {
delegate_->MessageReceived(session_id, message);
}
void InspectorIoDelegate::MessageReceived(int session_id,
const std::string& message) {
// 通过id找到对应的对象
auto session = sessions_.find(session_id);
if (session != sessions_.end())
session->second->Dispatch(Utf8ToStringView(message)->string());
}

sessions_里维护的是MainThreadSessionState对象，我们看MainThreadSessionState的Dispatch。

void Dispatch(std::unique_ptr<StringBuffer> message) {
session_->Dispatch(message->string());
}

MainThreadSessionState对象里的session_是SameThreadInspectorSession对象。

void SameThreadInspectorSession::Dispatch(
const v8_inspector::StringView& message) {
auto client = client_.lock();
if (client)
client->dispatchMessageFromFrontend(session_id_, message);
}

继续看dispatchMessageFromFrontend

void dispatchMessageFromFrontend(int session_id, const StringView& message) {
channels_[session_id]->dispatchProtocolMessage(message);
}

dispatchMessageFromFrontend通过session_id找到对应的channel，继续调用channel的dispatchProtocolMessage。

void dispatchProtocolMessage(const StringView& message) {
std::string raw_message = protocol::StringUtil::StringViewToUtf8(message);
// 解析数据
std::unique_ptr<protocol::DictionaryValue> value =
protocol::DictionaryValue::cast(protocol::StringUtil::parseMessage(
raw_message, false));
int call_id;
std::string method;
node_dispatcher_->parseCommand(value.get(), &call_id, &method);
if (v8_inspector::V8InspectorSession::canDispatchMethod(
Utf8ToStringView(method)->string())) {
// 把数据传给V8 inspector
session_->dispatchProtocolMessage(message);
} else {
node_dispatcher_->dispatch(call_id, method, std::move(value),
raw_message);
}
}

这里通过canDispatchMethod判断请求的指令是否是给V8 inspector的。

bool V8InspectorSession::canDispatchMethod(const StringView& method) {
return stringViewStartsWith(method,
protocol::Runtime::Metainfo::commandPrefix) ||
stringViewStartsWith(method,
protocol::Debugger::Metainfo::commandPrefix) ||
stringViewStartsWith(method,
protocol::Profiler::Metainfo::commandPrefix) ||
stringViewStartsWith(
method, protocol::HeapProfiler::Metainfo::commandPrefix) ||
stringViewStartsWith(method,
protocol::Console::Metainfo::commandPrefix) ||
stringViewStartsWith(method,
protocol::Schema::Metainfo::commandPrefix);
}

我们看到是通过前缀判断的，如果我们对文章开始的指令还有印象的话，就会知道大概有哪些指令。这就完成了数据从客户端到V8 inspector的传递。

3.2.3 V8 Inspector数据到客户端的处理流程

接着看从V8 inspector到客户端的数据传递逻辑。从前面分析我们知道V8 inspector通过channel的sendResponse函数传递输出的数据。

void sendResponse(
int callId,
std::unique_ptr<v8_inspector::StringBuffer> message) override {
sendMessageToFrontend(message->string());
}
void sendMessageToFrontend(const StringView& message) {
delegate_->SendMessageToFrontend(message);
}

delegate_是IoSessionDelegate对象。

void SendMessageToFrontend(const v8_inspector::StringView& message) override {
request_queue_->Post(id_, TransportAction::kSendMessage,
StringBuffer::create(message));
}

request_queue_是RequestQueueData对象。

void Post(int session_id,
TransportAction action,
std::unique_ptr<StringBuffer> message) {
Mutex::ScopedLock scoped_lock(state_lock_);
bool notify = messages_.empty();
messages_.emplace_back(action, session_id, std::move(message));
if (notify) {
CHECK_EQ(0, uv_async_send(&async_));
incoming_message_cond_.Broadcast(scoped_lock);
}
}

Post首先把消息入队，然后通过异步的方式通知async_接着看async_的处理函数(在子线程的事件循环里执行)。

int err = uv_async_init(loop, &async_, [](uv_async_t* async) {
// 拿到async对应的上下文
RequestQueueData* wrapper = node::ContainerOf(&RequestQueueData::async_, async);
// 执行RequestQueueData的DoDispatch
wrapper->DoDispatch();
});void DoDispatch() {
for (const auto& request : GetMessages()) {
request.Dispatch(server_);
}
}

request是RequestToServer对象。

void Dispatch(InspectorSocketServer* server) const {
switch (action_) {
case TransportAction::kSendMessage:
server->Send(
session_id_,
protocol::StringUtil::StringViewToUtf8(message_->string()));
break;
}
}

接着看InspectorSocketServer的Send。

void InspectorSocketServer::Send(int session_id, const std::string& message) {
SocketSession* session = Session(session_id);
if (session != nullptr) {
session->Send(message);
}
}

session代表可客户端的一个websocket连接。

void SocketSession::Send(const std::string& message) {
ws_socket_->Write(message.data(), message.length());}

接着调用websocket handler的Write。

void Write(const std::vector<char> data) override {
std::vector<char> output = encode_frame_hybi17(data);
WriteRaw(output, WriteRequest::Cleanup);
}

WriteRaw是基类ProtocolHandler实现的。

int ProtocolHandler::WriteRaw(const std::vector<char>& buffer,
uv_write_cb write_cb) {
return tcp_->WriteRaw(buffer, write_cb);
}

最终是通过TCP连接返回给客户端。

int TcpHolder::WriteRaw(const std::vector<char>& buffer, uv_write_cb write_cb) {
// Freed in write_request_cleanup
WriteRequest* wr = new WriteRequest(handler_, buffer);
uv_stream_t* stream = reinterpret_cast<uv_stream_t*>(&tcp_);
int err = uv_write(&wr->req, stream, &wr->buf, 1, write_cb);
if (err < 0)
delete wr;
return err < 0;
}

新建一个写请求，socket可写的时候发送数据给客户端。

4 动态开启Inspector

默认打开Inspector能力是不安全的，这意味着能连上websocket服务器的客户端都能通过协议控制Node.js进程，通常我们是在Node.js进程出现问题的时候，动态开启Inspector。

const http = require('http');const inspector = require('inspector');const fs = require('fs');
http.createServer((req, res) => {
if (req.url == 'debug') {
const session = new inspector.Session();
session.connect();
session.post('Profiler.enable', () => {
session.post('Profiler.start', () => {
session.post('Profiler.stop', (err, { profile }) => {
if (!err) {
fs.writeFileSync('./profile.cpuprofile', JSON.stringify(profile));
}
session.disconnect();
res.end('ok');
});
});
});
} else {
res.end('ok');
}
}).listen(80);