我们聊一下Node.jsInspector源码解析

之前的文章分析了Node.js Inspector的使用和原理，并粗略地分析了其源码，因为Node.js Inspector的实现非常复杂，逻辑又非常绕，所以本文打算更深入、更通俗地讲解Node.js Inspector的实现。

成都创新互联是创新、创意、研发型一体的综合型网站建设公司，自成立以来公司不断探索创新，始终坚持为客户提供满意周到的服务，在本地打下了良好的口碑，在过去的10年时间我们累计服务了上千家以及全国政企客户，如凿毛机等企业单位，完善的项目管理流程，严格把控项目进度与质量监控加上过硬的技术实力获得客户的一致表扬。

当我们以以下方式执行我们的应用时

 
 
 
 
    
  
  
  
  
  1. node inspect app.js 
    2.

1 初始化

Node.js在启动的过程中，就会初始化Inspector相关的逻辑。

 
 
 
 
    
  
  
  
  
  1. inspector_agent_ = std::make_unique(this); 
    2.

Agent是负责和V8 Inspector通信的对象。创建完后接着执行env->InitializeInspector({})启动Agent。

 
 
 
 
    
  
  
  
  
  1. inspector_agent_->Start(...); 
    2.

Start继续执行Agent::StartIoThread。

 
 
 
 
    
  
  
  
  
  1. bool Agent::StartIoThread() { 
    2. 
  
  
  
  
  2.   io_ = InspectorIo::Start(client_->getThreadHandle(), ...); 
    3. 
  
  
  
  
  3.   return true; 
    4. 
  
  
  
  
  4. } 
    5.

StartIoThread中的client_->getThreadHandle()是重要的逻辑，我们先来分析该函数。

 
 
 
 
    
  
  
  
  
  1. std::shared_ptr getThreadHandle() { 
    2. 
  
  
  
  
  2.     if (!interface_) { 
    3. 
  
  
  
  
  3.       interface_ = std::make_shared(env_->inspector_agent(), ...); 
    4. 
  
  
  
  
  4.     } 
    5. 
  
  
  
  
  5.     return interface_->GetHandle(); 
    6. 
  
  
  
  
  6. } 
    7.

getThreadHandle首先创建来一个MainThreadInterface对象，接着又调用了他的GetHandle方法，我们看一下该方法的逻辑。

 
 
 
 
    
  
  
  
  
  1. std::shared_ptr MainThreadInterface::GetHandle() { 
    2. 
  
  
  
  
  2.   if (handle_ == nullptr) 
    3. 
  
  
  
  
  3.     handle_ = std::make_shared(this); 
    4. 
  
  
  
  
  4.   return handle_; 
    5. 
  
  
  
  
  5. } 
    6.

GetHandlei了创建了一个MainThreadHandle对象，最终结构如下所示。

分析完后我们继续看Agent::StartIoThread中InspectorIo::Start的逻辑。

 
 
 
 
    
  
  
  
  
  1. std::unique_ptr InspectorIo::Start(std::shared_ptr main_thread, ...) { 
    2. 
  
  
  
  
  2.   auto io = std::unique_ptr(new InspectorIo(main_thread, ...)); 
    3. 
  
  
  
  
  3.   return io; 
    4. 
  
  
  
  
  4. } 
    5.

InspectorIo::Star里新建了一个InspectorIo对象，我们看看InspectorIo构造函数的逻辑。

 
 
 
 
    
  
  
  
  
  1. InspectorIo::InspectorIo(std::shared_ptr main_thread, ...) 
    2. 
  
  
  
  
  2.     :  
    3. 
  
  
  
  
  3.     // 初始化main_thread_ 
    4. 
  
  
  
  
  4.     main_thread_(main_thread)) { 
    5. 
  
  
  
  
  5.   // 新建一个子线程，子线程中执行InspectorIo::ThreadMain 
    6. 
  
  
  
  
  6.   uv_thread_create(&thread_, InspectorIo::ThreadMain, this); 
    7. 
  
  
  
  
  7. } 
    8.

这时候结构如下。

Inspector在子线程里启动的原因主要有两个。

1 如果在主线程里运行，那么当我们断点调试的时候，Node.js主线程就会被停住，也就无法处理客户端发过来的调试指令。

2 如果主线程陷入死循环，我们就无法实时抓取进程的profile数据来分析原因。接着继续看一下子线程里执行InspectorIo::ThreadMain的逻辑。

 
 
 
 
    
  
  
  
  
  1. void InspectorIo::ThreadMain(void* io) { 
    2. 
  
  
  
  
  2.   static_cast(io)->ThreadMain(); 
    3. 
  
  
  
  
  3. } 
    4. 
  
  
  
  
  4.  
    5. 
  
  
  
  
  5. void InspectorIo::ThreadMain() { 
    6. 
  
  
  
  
  6.   uv_loop_t loop; 
    7. 
  
  
  
  
  7.   loop.data = nullptr; 
    8. 
  
  
  
  
  8.   // 在子线程开启一个新的事件循环 
    9. 
  
  
  
  
  9.   int err = uv_loop_init(&loop); 
    10. 
  
  
  
  
  10.   std::shared_ptr queue(new RequestQueueData(&loop), ...); 
    11. 
  
  
  
  
  11.   // 新建一个delegate，用于处理请求 
    12. 
  
  
  
  
  12.   std::unique_ptr delegate( 
    13. 
  
  
  
  
  13.     new InspectorIoDelegate(queue, main_thread_, ...) 
    14. 
  
  
  
  
  14.   ); 
    15. 
  
  
  
  
  15.   InspectorSocketServer server(std::move(delegate), ...); 
    16. 
  
  
  
  
  16.   server.Start() 
    17. 
  
  
  
  
  17.   uv_run(&loop, UV_RUN_DEFAULT); 
    18. 
  
  
  
  
  18. } 
    19.

ThreadMain里主要三个逻辑

1 创建一个delegate对象，该对象是核心的对象，后面我们会看到有什么作用。

2 创建一个服务器并启动。

3 开启事件循环。接下来看一下服务器的逻辑，首先看一下创建服务器的逻辑。

 
 
 
 
    
  
  
  
  
  1. InspectorSocketServer::InspectorSocketServer(std::unique_ptr delegate, ...) 
    2. 
  
  
  
  
  2.     :  
    3. 
  
  
  
  
  3.       // 保存delegate 
    4. 
  
  
  
  
  4.       delegate_(std::move(delegate)), 
    5. 
  
  
  
  
  5.       // 初始化sessionId 
    6. 
  
  
  
  
  6.       next_session_id_(0) { 
    7. 
  
  
  
  
  7.   // 设置delegate的server为当前服务器 
    8. 
  
  
  
  
  8.   delegate_->AssignServer(this); 
    9. 
  
  
  
  
  9. } 
    10.

执行完后形成以下结构。

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

 
 
 
 
    
  
  
  
  
  1. bool InspectorSocketServer::Start() { 
    2. 
  
  
  
  
  2.   // DNS解析,比如输入的是localhost 
    3. 
  
  
  
  
  3.   struct addrinfo hints; 
    4. 
  
  
  
  
  4.   memset(&hints, 0, sizeof(hints)); 
    5. 
  
  
  
  
  5.   hints.ai_flags = AI_NUMERICSERV; 
    6. 
  
  
  
  
  6.   hints.ai_socktype = SOCK_STREAM; 
    7. 
  
  
  
  
  7.   uv_getaddrinfo_t req; 
    8. 
  
  
  
  
  8.   const std::string port_string = std::to_string(port_); 
    9. 
  
  
  
  
  9.   uv_getaddrinfo(loop_, &req, nullptr, host_.c_str(), 
    10. 
  
  
  
  
  10.                            port_string.c_str(), &hints); 
    11. 
  
  
  
  
  11.   // 监听解析到的ip列表                  
    12. 
  
  
  
  
  12.   for (addrinfo* address = req.addrinfo;  
    13. 
  
  
  
  
  13.        address != nullptr; 
    14. 
  
  
  
  
  14.        address = address->ai_next) { 
    15. 
  
  
  
  
  15.  
    16. 
  
  
  
  
  16.     auto server_socket = ServerSocketPtr(new ServerSocket(this)); 
    17. 
  
  
  
  
  17.     err = server_socket->Listen(address->ai_addr, loop_); 
    18. 
  
  
  
  
  18.     if (err == 0) 
    19. 
  
  
  
  
  19.       server_sockets_.push_back(std::move(server_socket)); 
    20. 
  
  
  
  
  20.  
    21. 
  
  
  
  
  21.   } 
    22. 
  
  
  
  
  22.  
    23. 
  
  
  
  
  23.   return true; 
    24. 
  
  
  
  
  24. } 
    25.

首先根据参数做一个DNS解析，然后根据拿到的ip列表(通常是一个)，创建对应个数的ServerSocket对象，并执行他的Listen方法。ServerSocket表示一个监听socket。看一下ServerSocket的构造函数。

 
 
 
 
    
  
  
  
  
  1. ServerSocket(InspectorSocketServer* server) 
    2. 
  
  
  
  
  2.             : tcp_socket_(uv_tcp_t()), server_(server) {} 
    3.

执行完后结构如下。

接着看一下ServerSocket的Listen方法。

 
 
 
 
    
  
  
  
  
  1. int ServerSocket::Listen(sockaddr* addr, uv_loop_t* loop) { 
    2. 
  
  
  
  
  2.   uv_tcp_t* server = &tcp_socket_; 
    3. 
  
  
  
  
  3.   uv_tcp_init(loop, server) 
    4. 
  
  
  
  
  4.   uv_tcp_bind(server, addr, 0); 
    5. 
  
  
  
  
  5.   uv_listen(reinterpret_cast(server),  
    6. 
  
  
  
  
  6.                     511, 
    7. 
  
  
  
  
  7.                     ServerSocket::SocketConnectedCallback); 
    8. 
  
  
  
  
  8. } 
    9.

Listen调用Libuv的接口完成服务器的启动。至此，Inspector提供的Weboscket服务器启动了。

2 处理连接

从刚才分析中可以看到，当有连接到来时执行回调ServerSocket::SocketConnectedCallback。

 
 
 
 
    
  
  
  
  
  1. void ServerSocket::SocketConnectedCallback(uv_stream_t* tcp_socket, 
    2. 
  
  
  
  
  2.                                            int status) { 
    3. 
  
  
  
  
  3.   if (status == 0) { 
    4. 
  
  
  
  
  4.     // 根据Libuv handle找到对应的ServerSocket对象 
    5. 
  
  
  
  
  5.     ServerSocket* server_socket = ServerSocket::FromTcpSocket(tcp_socket); 
    6. 
  
  
  
  
  6.     // Socket对象的server_字段保存了所在的InspectorSocketServer 
    7. 
  
  
  
  
  7.     server_socket->server_->Accept(server_socket->port_, tcp_socket); 
    8. 
  
  
  
  
  8.   } 
    9. 
  
  
  
  
  9. } 
    10.

接着看InspectorSocketServer的Accept是如何处理连接的。

 
 
 
 
    
  
  
  
  
  1. void InspectorSocketServer::Accept(int server_port, 
    2. 
  
  
  
  
  2.                                    uv_stream_t* server_socket) { 
    3. 
  
  
  
  
  3.  
    4. 
  
  
  
  
  4.   std::unique_ptr session( 
    5. 
  
  
  
  
  5.       new SocketSession(this, next_session_id_++, server_port) 
    6. 
  
  
  
  
  6.   ); 
    7. 
  
  
  
  
  7.  
    8. 
  
  
  
  
  8.   InspectorSocket::DelegatePointer delegate = 
    9. 
  
  
  
  
  9.       InspectorSocket::DelegatePointer( 
    10. 
  
  
  
  
  10.           new SocketSession::Delegate(this, session->id()) 
    11. 
  
  
  
  
  11.       ); 
    12. 
  
  
  
  
  12.  
    13. 
  
  
  
  
  13.   InspectorSocket::Pointer inspector = 
    14. 
  
  
  
  
  14.       InspectorSocket::Accept(server_socket, std::move(delegate)); 
    15. 
  
  
  
  
  15.  
    16. 
  
  
  
  
  16.   if (inspector) { 
    17. 
  
  
  
  
  17.     session->Own(std::move(inspector)); 
    18. 
  
  
  
  
  18.     connected_sessions_[session->id()].second = std::move(session); 
    19. 
  
  
  
  
  19.   } 
    20. 
  
  
  
  
  20. } 
    21.

Accept的首先创建里一个SocketSession和SocketSession::Delegate对象。然后调用InspectorSocket::Accept，从代码中可以看到InspectorSocket::Accept会返回一个InspectorSocket对象。InspectorSocket是对通信socket的封装(和客户端通信的socket，区别于服务器的监听socket)。然后记录session对象对应的InspectorSocket对象，同时记录sessionId和session的映射关系。结构如下图所示。

接着看一下InspectorSocket::Accept返回InspectorSocket的逻辑。

 
 
 
 
    
  
  
  
  
  1. InspectorSocket::Pointer InspectorSocket::Accept(uv_stream_t* server, 
    2. 
  
  
  
  
  2.                                                  DelegatePointer delegate) { 
    3. 
  
  
  
  
  3.   auto tcp = TcpHolder::Accept(server, std::move(delegate)); 
    4. 
  
  
  
  
  4.   InspectorSocket* inspector = new InspectorSocket(); 
    5. 
  
  
  
  
  5.   inspector->SwitchProtocol(new HttpHandler(inspector, std::move(tcp))); 
    6. 
  
  
  
  
  6.   return InspectorSocket::Pointer(inspector); 
    7. 
  
  
  
  
  7. } 
    8.

InspectorSocket::Accept的代码不多，但是逻辑还是挺多的。

 
 
 
 
    
  
  
  
  
  1. TcpHolder::Pointer TcpHolder::Accept( 
    2. 
  
  
  
  
  2.     uv_stream_t* server, 
    3. 
  
  
  
  
  3.     InspectorSocket::DelegatePointer delegate) { 
    4. 
  
  
  
  
  4.   // 新建一个TcpHolder对象，TcpHolder是对uv_tcp_t和delegate的封装 
    5. 
  
  
  
  
  5.   TcpHolder* result = new TcpHolder(std::move(delegate)); 
    6. 
  
  
  
  
  6.   // 拿到TcpHolder对象的uv_tcp_t结构体 
    7. 
  
  
  
  
  7.   uv_stream_t* tcp = reinterpret_cast(&result->tcp_); 
    8. 
  
  
  
  
  8.   // 初始化 
    9. 
  
  
  
  
  9.   int err = uv_tcp_init(server->loop, &result->tcp_); 
    10. 
  
  
  
  
  10.   // 摘取一个TCP连接对应的fd保存到TcpHolder的uv_tcp_t结构体中（即第二个参数的tcp字段） 
    11. 
  
  
  
  
  11.   uv_accept(server, tcp); 
    12. 
  
  
  
  
  12.   // 注册等待可读事件，有数据时执行OnDataReceivedCb回调 
    13. 
  
  
  
  
  13.   uv_read_start(tcp, allocate_buffer, OnDataReceivedCb); 
    14. 
  
  
  
  
  14.   return TcpHolder::Pointer(result); 
    15. 
  
  
  
  
  15. } 
    16.

2 新建一个HttpHandler对象。

 
 
 
 
    
  
  
  
  
  1. explicit HttpHandler(InspectorSocket* inspector, TcpHolder::Pointer tcp) 
    2. 
  
  
  
  
  2.                      : ProtocolHandler(inspector, std::move(tcp)){ 
    3. 
  
  
  
  
  3.  
    4. 
  
  
  
  
  4.   llhttp_init(&parser_, HTTP_REQUEST, &parser_settings); 
    5. 
  
  
  
  
  5.   llhttp_settings_init(&parser_settings); 
    6. 
  
  
  
  
  6.   parser_settings.on_header_field = OnHeaderField; 
    7. 
  
  
  
  
  7.   parser_settings.on_header_value = OnHeaderValue; 
    8. 
  
  
  
  
  8.   parser_settings.on_message_complete = OnMessageComplete; 
    9. 
  
  
  
  
  9.   parser_settings.on_url = OnPath; 
    10. 
  
  
  
  
  10. } 
    11. 
  
  
  
  
  11.  
    12. 
  
  
  
  
  12. ProtocolHandler::ProtocolHandler(InspectorSocket* inspector, 
    13. 
  
  
  
  
  13.                                  TcpHolder::Pointer tcp) 
    14. 
  
  
  
  
  14.                                  : inspector_(inspector), tcp_(std::move(tcp)) { 
    15. 
  
  
  
  
  15.   // 设置TCP数据的handler，TCP是只负责传输，数据的解析交给handler处理                                
    16. 
  
  
  
  
  16.   tcp_->SetHandler(this); 
    17. 
  
  
  
  
  17. } 
    18.

HttpHandler是对uv_tcp_t的封装，主要通过HTTP解析器llhttp对HTTP协议进行解析。

3 调用inspector->SwitchProtocol()切换当前协议为HTTP，建立TCP连接后，首先要经过一个HTTP请求从HTTP协议升级到WebSocket协议，升级成功后就使用Websocket协议进行通信。我们看一下这时候的结构图。

至此，就完成了连接处理的分析。

3 协议升级

完成了TCP连接的处理后，接下来要完成协议升级，因为Inspector是通过WebSocket协议和客户端通信的，所以需要通过一个HTTP请求来完成HTTP到WebSocekt协议的升级。从刚才的分析中看当有数据到来时会执行OnDataReceivedCb回调。

 
 
 
 
    
  
  
  
  
  1. void TcpHolder::OnDataReceivedCb(uv_stream_t* tcp, ssize_t nread, 
    2. 
  
  
  
  
  2.                                  const uv_buf_t* buf) { 
    3. 
  
  
  
  
  3.   TcpHolder* holder = From(tcp); 
    4. 
  
  
  
  
  4.   holder->ReclaimUvBuf(buf, nread); 
    5. 
  
  
  
  
  5.   // 调用handler的onData，目前handler是HTTP协议 
    6. 
  
  
  
  
  6.   holder->handler_->OnData(&holder->buffer); 
    7. 
  
  
  
  
  7. } 
    8.

TCP层收到数据后交给应用层解析，直接调用上层的OnData回调。

 
 
 
 
    
  
  
  
  
  1. void OnData(std::vector* data) override { 
    2. 
  
  
  
  
  2.     // 解析HTTP协议 
    3. 
  
  
  
  
  3.     llhttp_execute(&parser_, data->data(), data->size()); 
    4. 
  
  
  
  
  4.     // 解析完并且是升级协议的请求则调用delegate的回调OnSocketUpgrade 
    5. 
  
  
  
  
  5.     delegate()->OnSocketUpgrade(event.host, event.path, event.ws_key); 
    6. 
  
  
  
  
  6. } 
    7.

OnData可能会被多次回调，并通过llhttp_execute解析收到的HTTP报文，当发现是一个协议升级的请求后，就调用OnSocketUpgrade回调。delegate是TCP层保存的SocketSession::Delegate对象。来看一下该对象的OnSocketUpgrade方法。

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

OnSocketUpgrade又调用来server_(InspectorSocketServer对象)的SessionStarted。

 
 
 
 
    
  
  
  
  
  1. void InspectorSocketServer::SessionStarted(int session_id, 
    2. 
  
  
  
  
  2.                                            const std::string& id, 
    3. 
  
  
  
  
  3.                                            const std::string& ws_key) { 
    4. 
  
  
  
  
  4.   // 找到对应的session对象                                            
    5. 
  
  
  
  
  5.   SocketSession* session = Session(session_id); 
    6. 
  
  
  
  
  6.   connected_sessions_[session_id].first = id; 
    7. 
  
  
  
  
  7.   session->Accept(ws_key); 
    8. 
  
  
  
  
  8.   delegate_->StartSession(session_id, id); 
    9. 
  
  
  
  
  9. } 
    10.

首先通过session_id找到建立TCP连接时分配的SocketSession对象。

1 执行session->Accept(ws_key);回复客户端同意协议升级。

 
 
 
 
    
  
  
  
  
  1. void Accept(const std::string& ws_key) { 
    2. 
  
  
  
  
  2.   ws_socket_->AcceptUpgrade(ws_key); 
    3. 
  
  
  
  
  3. } 
    4.

从结构图我们可以看到ws_socket_是一个InspectorSocket对象。

 
 
 
 
    
  
  
  
  
  1. void AcceptUpgrade(const std::string& accept_key) override { 
    2. 
  
  
  
  
  2.     char accept_string[ACCEPT_KEY_LENGTH]; 
    3. 
  
  
  
  
  3.     generate_accept_string(accept_key, &accept_string); 
    4. 
  
  
  
  
  4.     const char accept_ws_prefix[] = "HTTP/1.1 101 Switching Protocols\r\n" 
    5. 
  
  
  
  
  5.                                     "Upgrade: websocket\r\n" 
    6. 
  
  
  
  
  6.                                     "Connection: Upgrade\r\n" 
    7. 
  
  
  
  
  7.                                     "Sec-WebSocket-Accept: "; 
    8. 
  
  
  
  
  8.     const char accept_ws_suffix[] = "\r\n\r\n"; 
    9. 
  
  
  
  
  9.     std::vector reply(accept_ws_prefix, 
    10. 
  
  
  
  
  10.                             accept_ws_prefix + sizeof(accept_ws_prefix) - 1); 
    11. 
  
  
  
  
  11.     reply.insert(reply.end(), accept_string, 
    12. 
  
  
  
  
  12.                  accept_string + sizeof(accept_string)); 
    13. 
  
  
  
  
  13.     reply.insert(reply.end(), accept_ws_suffix, 
    14. 
  
  
  
  
  14.                  accept_ws_suffix + sizeof(accept_ws_suffix) - 1); 
    15. 
  
  
  
  
  15.     // 回复101给客户端              
    16. 
  
  
  
  
  16.     WriteRaw(reply, WriteRequest::Cleanup); 
    17. 
  
  
  
  
  17.     // 切换handler为WebSocket handler 
    18. 
  
  
  
  
  18.     inspector_->SwitchProtocol(new WsHandler(inspector_, std::move(tcp_))); 
    19. 
  
  
  
  
  19. } 
    20.

AcceptUpgradeh首先回复客户端101表示同意升级道WebSocket协议，然后切换数据处理器为WsHandler，即后续的数据按照WebSocket协议处理。

2 执行delegate_->StartSession(session_id, id)建立和V8 Inspector的会话。delegate_是InspectorIoDelegate对象。

 
 
 
 
    
  
  
  
  
  1. void InspectorIoDelegate::StartSession(int session_id, 
    2. 
  
  
  
  
  2.                                        const std::string& target_id) { 
    3. 
  
  
  
  
  3.   auto session = main_thread_->Connect( 
    4. 
  
  
  
  
  4.       std::unique_ptr( 
    5. 
  
  
  
  
  5.           new IoSessionDelegate(request_queue_->handle(), session_id) 
    6. 
  
  
  
  
  6.       ),  
    7. 
  
  
  
  
  7.       true); 
    8. 
  
  
  
  
  8.   if (session) { 
    9. 
  
  
  
  
  9.     sessions_[session_id] = std::move(session); 
    10. 
  
  
  
  
  10.     fprintf(stderr, "Debugger attached.\n"); 
    11. 
  
  
  
  
  11.   } 
    12. 
  
  
  
  
  12. } 
    13.

首先通过main_thread_->Connect拿到一个session，并在InspectorIoDelegate中记录映射关系。结构图如下。

接下来看一下main_thread_->Connect的逻辑(main_thread_是MainThreadHandle对象)。

 
 
 
 
    
  
  
  
  
  1. std::unique_ptr MainThreadHandle::Connect( 
    2. 
  
  
  
  
  2.     std::unique_ptr delegate, 
    3. 
  
  
  
  
  3.     bool prevent_shutdown) { 
    4. 
  
  
  
  
  4.  
    5. 
  
  
  
  
  5.   return std::unique_ptr( 
    6. 
  
  
  
  
  6.       new CrossThreadInspectorSession(++next_session_id_, 
    7. 
  
  
  
  
  7.                                       shared_from_this(), 
    8. 
  
  
  
  
  8.                                       std::move(delegate), 
    9. 
  
  
  
  
  9.                                       prevent_shutdown)); 
    10. 
  
  
  
  
  10. } 
    11.

Connect函数新建了一个CrossThreadInspectorSession对象。

 
 
 
 
    
  
  
  
  
  1. CrossThreadInspectorSession( 
    2. 
  
  
  
  
  2.       int id, 
    3. 
  
  
  
  
  3.       std::shared_ptr thread, 
    4. 
  
  
  
  
  4.       std::unique_ptr delegate, 
    5. 
  
  
  
  
  5.       bool prevent_shutdown) 
    6. 
  
  
  
  
  6.       // 创建一个MainThreadSessionState对象 
    7. 
  
  
  
  
  7.       : state_(thread, std::bind(MainThreadSessionState::Create, 
    8. 
  
  
  
  
  8.                                  std::placeholders::_1, 
    9. 
  
  
  
  
  9.                                  prevent_shutdown)) { 
    10. 
  
  
  
  
  10.     // 执行MainThreadSessionState::Connect                              
    11. 
  
  
  
  
  11.     state_.Call(&MainThreadSessionState::Connect, std::move(delegate)); 
    12. 
  
  
  
  
  12.   } 
    13.

继续看MainThreadSessionState::Connect。

 
 
 
 
    
  
  
  
  
  1. void Connect(std::unique_ptr delegate) { 
    2. 
  
  
  
  
  2.     Agent* agent = thread_->inspector_agent(); 
    3. 
  
  
  
  
  3.     session_ = agent->Connect(std::move(delegate), prevent_shutdown_); 
    4. 
  
  
  
  
  4. } 
    5.

继续调agent->Connect。

 
 
 
 
    
  
  
  
  
  1. std::unique_ptr Agent::Connect( 
    2. 
  
  
  
  
  2.     std::unique_ptr delegate, 
    3. 
  
  
  
  
  3.     bool prevent_shutdown) { 
    4. 
  
  
  
  
  4.  
    5. 
  
  
  
  
  5.   int session_id = client_->connectFrontend(std::move(delegate), 
    6. 
  
  
  
  
  6.                                             prevent_shutdown); 
    7. 
  
  
  
  
  7.   return std::unique_ptr( 
    8. 
  
  
  
  
  8.       new SameThreadInspectorSession(session_id, client_)); 
    9. 
  
  
  
  
  9. } 
    10.

继续调connectFrontend

 
 
 
 
    
  
  
  
  
  1. int connectFrontend(std::unique_ptr delegate, 
    2. 
  
  
  
  
  2.                       bool prevent_shutdown) { 
    3. 
  
  
  
  
  3.     int session_id = next_session_id_++; 
    4. 
  
  
  
  
  4.     channels_[session_id] = std::make_unique(env_, 
    5. 
  
  
  
  
  5.                                                           client_, 
    6. 
  
  
  
  
  6.                                                           getWorkerManager(), 
    7. 
  
  
  
  
  7.                                                           std::move(delegate), 
    8. 
  
  
  
  
  8.                                                           getThreadHandle(), 
    9. 
  
  
  
  
  9.                                                           prevent_shutdown); 
    10. 
  
  
  
  
  10.     return session_id; 
    11. 
  
  
  
  
  11. } 
    12.

connectFrontend创建了一个ChannelImpl并且在channels_中保存了映射关系。看看ChannelImpl的构造函数。

 
 
 
 
    
  
  
  
  
  1. explicit ChannelImpl(Environment* env, 
    2. 
  
  
  
  
  2.                      const std::unique_ptr& inspector, 
    3. 
  
  
  
  
  3.                      std::unique_ptr delegate, ...) 
    4. 
  
  
  
  
  4.       : delegate_(std::move(delegate)) { 
    5. 
  
  
  
  
  5.  
    6. 
  
  
  
  
  6.     session_ = inspector->connect(CONTEXT_GROUP_ID, this, StringView()); 
    7. 
  
  
  
  
  7. } 
    8.

ChannelImpl调用inspector->connect建立了一个和V8 Inspector的会话。结构图大致如下。

4 客户端到V8 Inspector的数据处理

TCP连接建立了，协议升级也完成了，接下来就可以开始处理业务数据。从前面的分析中我们已经知道数据到来时会执行TcpHoldler的handler_->OnData回调。因为已经完成了协议升级，所以这时候的handler变成了WeSocket handler。

 
 
 
 
    
  
  
  
  
  1. void OnData(std::vector* data) override { 
    2. 
  
  
  
  
  2.     // 1. Parse. 
    3. 
  
  
  
  
  3.     int processed = 0; 
    4. 
  
  
  
  
  4.     do { 
    5. 
  
  
  
  
  5.       processed = ParseWsFrames(*data); 
    6. 
  
  
  
  
  6.       // 2. Fix the data size & length 
    7. 
  
  
  
  
  7.       if (processed > 0) { 
    8. 
  
  
  
  
  8.         remove_from_beginning(data, processed); 
    9. 
  
  
  
  
  9.       } 
    10. 
  
  
  
  
  10.     } while (processed > 0 && !data->empty()); 
    11. 
  
  
  
  
  11. } 
    12.

OnData通过ParseWsFrames解析WebSocket协议。

 
 
 
 
    
  
  
  
  
  1. int ParseWsFrames(const std::vector& buffer) { 
    2. 
  
  
  
  
  2.     int bytes_consumed = 0; 
    3. 
  
  
  
  
  3.     std::vector output; 
    4. 
  
  
  
  
  4.     bool compressed = false; 
    5. 
  
  
  
  
  5.     // 解析WebSocket协议 
    6. 
  
  
  
  
  6.     ws_decode_result r =  decode_frame_hybi17(buffer, 
    7. 
  
  
  
  
  7.                                               true /* client_frame */, 
    8. 
  
  
  
  
  8.                                               &bytes_consumed, &output, 
    9. 
  
  
  
  
  9.                                               &compressed); 
    10. 
  
  
  
  
  10.     // 执行delegate的回调                                         
    11. 
  
  
  
  
  11.     delegate()->OnWsFrame(output); 
    12. 
  
  
  
  
  12.     return bytes_consumed; 
    13. 
  
  
  
  
  13. } 
    14.

前面已经分析过delegate是TcpHoldler的delegate，即SocketSession::Delegate对象。

 
 
 
 
    
  
  
  
  
  1. void SocketSession::Delegate::OnWsFrame(const std::vector& data) { 
    2. 
  
  
  
  
  2.   server_->MessageReceived(session_id_, 
    3. 
  
  
  
  
  3.                            std::string(data.data(),  
    4. 
  
  
  
  
  4.                            data.size())); 
    5. 
  
  
  
  
  5. } 
    6.

继续回调server_->MessageReceived。从结构图可以看到server_是InspectorSocketServer对象。

 
 
 
 
    
  
  
  
  
  1. void MessageReceived(int session_id, const std::string& message) { 
    2. 
  
  
  
  
  2.   delegate_->MessageReceived(session_id, message); 
    3. 
  
  
  
  
  3. } 
    4.

继续回调delegate_->MessageReceived。InspectorSocketServer的delegate_是InspectorIoDelegate对象。

 
 
 
 
    
  
  
  
  
  1. void InspectorIoDelegate::MessageReceived(int session_id, 
    2. 
  
  
  
  
  2.                                           const std::string& message) { 
    3. 
  
  
  
  
  3.   auto session = sessions_.find(session_id); 
    4. 
  
  
  
  
  4.   if (session != sessions_.end()) 
    5. 
  
  
  
  
  5.     session->second->Dispatch(Utf8ToStringView(message)->string()); 
    6. 
  
  
  
  
  6. } 
    7.

首先通过session_id找到对应的session。session是一个CrossThreadInspectorSession对象。看看他的Dispatch方法。

 
 
 
 
    
  
  
  
  
  1. void Dispatch(const StringView& message) override { 
    2. 
  
  
  
  
  2.     state_.Call(&MainThreadSessionState::Dispatch, 
    3. 
  
  
  
  
  3.                 StringBuffer::create(message)); 
    4. 
  
  
  
  
  4. } 
    5.

执行MainThreadSessionState::Dispatch。

 
 
 
 
    
  
  
  
  
  1. void Dispatch(std::unique_ptr message) { 
    2. 
  
  
  
  
  2.   session_->Dispatch(message->string()); 
    3. 
  
  
  
  
  3. } 
    4.

session_是SameThreadInspectorSession对象。

 
 
 
 
    
  
  
  
  
  1. void SameThreadInspectorSession::Dispatch( 
    2. 
  
  
  
  
  2.     const v8_inspector::StringView& message) { 
    3. 
  
  
  
  
  3.   auto client = client_.lock(); 
    4. 
  
  
  
  
  4.   if (client) 
    5. 
  
  
  
  
  5.     client->dispatchMessageFromFrontend(session_id_, message); 
    6. 
  
  
  
  
  6. } 
    7.

继续调client->dispatchMessageFromFrontend。

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

通过session_id找到对应的ChannelImpl，继续调ChannelImpl的dispatchProtocolMessage。

 
 
 
 
    
  
  
  
  
  1. voiddispatchProtocolMessage(const StringView& message) { 
    2. 
  
  
  
  
  2.    session_->dispatchProtocolMessage(message); 
    3. 
  
  
  
  
  3. } 
    4.

最终调用和V8 Inspector的会话对象把数据发送给V8。至此客户端到V8 Inspector的通信过程就完成了。

5 V8 Inspector到客户端的数据处理

接着看从V8 inspector到客户端的数据传递逻辑。V8 inspector是通过channel的sendResponse函数传递给客户端的。

 
 
 
 
    
  
  
  
  
  1. void sendResponse( 
    2. 
  
  
  
  
  2.       int callId, 
    3. 
  
  
  
  
  3.       std::unique_ptr message) override { 
    4. 
  
  
  
  
  4.  
    5. 
  
  
  
  
  5.     sendMessageToFrontend(message->string()); 
    6. 
  
  
  
  
  6.   } 
    7. 
  
  
  
  
  7.  
    8. 
  
  
  
  
  8.  void sendMessageToFrontend(const StringView& message) { 
    9. 
  
  
  
  
  9.     delegate_->SendMessageToFrontend(message); 
    10. 
  
  
  
  
  10.  } 
    11.

delegate_是IoSessionDelegate对象。

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

request_queue_是RequestQueueData对象。

 
 
 
 
    
  
  
  
  
  1. void Post(int session_id, 
    2. 
  
  
  
  
  2.             TransportAction action, 
    3. 
  
  
  
  
  3.             std::unique_ptr message) { 
    4. 
  
  
  
  
  4.  
    5. 
  
  
  
  
  5.     Mutex::ScopedLock scoped_lock(state_lock_); 
    6. 
  
  
  
  
  6.     bool notify = messages_.empty(); 
    7. 
  
  
  
  
  7.     messages_.emplace_back(action, session_id, std::move(message)); 
    8. 
  
  
  
  
  8.     if (notify) { 
    9. 
  
  
  
  
  9.       CHECK_EQ(0, uv_async_send(&async_)); 
    10. 
  
  
  
  
  10.       incoming_message_cond_.Broadcast(scoped_lock); 
    11. 
  
  
  
  
  11.     } 
    12. 
  
  
  
  
  12.   } 
    13.

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

 
 
 
 
    
  
  
  
  
  1. uv_async_init(loop, &async_, [](uv_async_t* async) { 
    2. 
  
  
  
  
  2.    // 拿到async对应的上下文 
    3. 
  
  
  
  
  3.    RequestQueueData* wrapper = node::ContainerOf(&RequestQueueData::async_, async); 
    4. 
  
  
  
  
  4.    // 执行RequestQueueData的DoDispatch 
    5. 
  
  
  
  
  5.    wrapper->DoDispatch();});void DoDispatch() { 
    6. 
  
  
  
  
  6.     for (const auto& request : GetMessages()) { 
    7. 
  
  
  
  
  7.       request.Dispatch(server_); 
    8. 
  
  
  
  
  8.     } 
    9. 
  
  
  
  
  9.   } 
    10.

request是RequestToServer对象。

 
 
 
 
    
  
  
  
  
  1. void Dispatch(InspectorSocketServer* server) const { 
    2. 
  
  
  
  
  2.     switch (action_) { 
    3. 
  
  
  
  
  3.       case TransportAction::kSendMessage: 
    4. 
  
  
  
  
  4.         server->Send( 
    5. 
  
  
  
  
  5.             session_id_, 
    6. 
  
  
  
  
  6.             protocol::StringUtil::StringViewToUtf8(message_->string())); 
    7. 
  
  
  
  
  7.         break; 
    8. 
  
  
  
  
  8.     } 
    9. 
  
  
  
  
  9.   } 
    10.

接着看InspectorSocketServer的Send。

 
 
 
 
    
  
  
  
  
  1. void InspectorSocketServer::Send(int session_id, const std::string& message) { 
    2. 
  
  
  
  
  2.   SocketSession* session = Session(session_id); 
    3. 
  
  
  
  
  3.   if (session != nullptr) { 
    4. 
  
  
  
  
  4.     session->Send(message); 
    5. 
  
  
  
  
  5.   } 
    6. 
  
  
  
  
  6. } 
    7.

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

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

接着调用WebSocket handler的Write。

 
 
 
 
    
  
  
  
  
  1. void Write(const std::vector data) override { 
    2. 
  
  
  
  
  2.     std::vector output = encode_frame_hybi17(data); 
    3. 
  
  
  
  
  3.     WriteRaw(output, WriteRequest::Cleanup); 
    4. 
  
  
  
  
  4.   } 
    5.

WriteRaw是基类ProtocolHandler实现的。

 
 
 
 
    
  
  
  
  
  1. int ProtocolHandler::WriteRaw(const std::vector& buffer, 
    2. 
  
  
  
  
  2.                               uv_write_cb write_cb) { 
    3. 
  
  
  
  
  3.   return tcp_->WriteRaw(buffer, write_cb); 
    4. 
  
  
  
  
  4. } 
    5.

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

 
 
 
 
    
  
  
  
  
  1. int TcpHolder::WriteRaw(const std::vector& buffer, uv_write_cb write_cb) { 
    2. 
  
  
  
  
  2.   // Freed in write_request_cleanup 
    3. 
  
  
  
  
  3.   WriteRequest* wr = new WriteRequest(handler_, buffer); 
    4. 
  
  
  
  
  4.   uv_stream_t* stream = reinterpret_cast(&tcp_); 
    5. 
  
  
  
  
  5.   int err = uv_write(&wr->req, stream, &wr->buf, 1, write_cb); 
    6. 
  
  
  
  
  6.   if (err < 0) 
    7. 
  
  
  
  
  7.     delete wr; 
    8. 
  
  
  
  
  8.   return err < 0; 
    9. 
  
  
  
  
  9. } 
    10.

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

后记：Node.js Inspector的原理虽然不复杂的，但是实现实在太绕了。

文章题目：我们聊一下Node.jsInspector源码解析
链接地址：http://www.gawzjz.com/qtweb2/news32/29732.html

网站建设、网络推广公司-创新互联，是专注品牌与效果的网站制作，网络营销seo公司；服务项目有等

广告

声明：本网站发布的内容（图片、视频和文字）以用户投稿、用户转载内容为主，如果涉及侵权请尽快告知，我们将会在第一时间删除。文章观点不代表本网站立场，如需处理请联系客服。电话：028-86922220；邮箱：631063699@qq.com。内容未经允许不得转载，或转载时需注明来源： 创新互联

猜你还喜欢下面的内容

品牌网站建设知识

同城分类信息

贵安服务器托管　　 　商标注册　　 　企业网站维护　　 　云服务器　　 　定制网站制作　　 　企业网站设计　　 　成都企业网站建设　　 　成都网站建设　　 　成都网站排名　　 　短视频拍摄　　 　成都网站建设　　 　成都网站设计　　 　眉山电信机房　　 　Android开发　　 　软文营销　　 　温江电信机房　　 　网站托管运营　　 　消防工程　　 　做网站价格　　 　内江托管服务器