| 1416 | } |
| 1417 | |
| 1418 | void ManagerService::DispatchPipeData(VmRuntimeHandle& rt, const std::string& vm_id) { |
| 1419 | while (!rt.recv_pending.empty()) { |
| 1420 | if (rt.recv_payload_needed > 0) { |
| 1421 | if (rt.recv_pending.size() < rt.recv_payload_needed) return; |
| 1422 | rt.recv_pending_msg.payload.assign( |
| 1423 | reinterpret_cast<const uint8_t*>(rt.recv_pending.data()), |
| 1424 | reinterpret_cast<const uint8_t*>(rt.recv_pending.data()) + rt.recv_payload_needed); |
| 1425 | rt.recv_pending.erase(0, rt.recv_payload_needed); |
| 1426 | rt.recv_payload_needed = 0; |
| 1427 | HandleIncomingMessage(vm_id, rt.recv_pending_msg); |
| 1428 | continue; |
| 1429 | } |
| 1430 | |
| 1431 | size_t nl = rt.recv_pending.find('\n'); |
| 1432 | if (nl == std::string::npos) return; |
| 1433 | std::string line = rt.recv_pending.substr(0, nl + 1); |
| 1434 | rt.recv_pending.erase(0, nl + 1); |
| 1435 | auto decoded = ipc::Decode(line); |
| 1436 | if (!decoded) continue; |
| 1437 | |
| 1438 | auto ps_it = decoded->fields.find("payload_size"); |
| 1439 | if (ps_it != decoded->fields.end()) { |
| 1440 | rt.recv_payload_needed = std::strtoull(ps_it->second.c_str(), nullptr, 10); |
| 1441 | decoded->fields.erase(ps_it); |
| 1442 | if (rt.recv_payload_needed > 0) { |
| 1443 | rt.recv_pending_msg = std::move(*decoded); |
| 1444 | continue; |
| 1445 | } |
| 1446 | } |
| 1447 | HandleIncomingMessage(vm_id, *decoded); |
| 1448 | } |
| 1449 | } |
| 1450 | |
| 1451 | void ManagerService::HandleProcessExit(const std::string& vm_id) { |
| 1452 | // Called from the libuv read callback on the loop thread. |
no test coverage detected