@brief send chunk data @param data the data @param is_final the is final
| 427 | ///@param data the data |
| 428 | ///@param is_final the is final |
| 429 | void HttpSession::send_chunk_data(const json& data, bool is_final) { |
| 430 | std::string chunk_content; |
| 431 | |
| 432 | // Check if the data is a string (pre-formatted SSE with "data: " prefix) |
| 433 | if (data.is_string()) { |
| 434 | std::string data_str = data.get<std::string>(); |
| 435 | // If it starts with "data: ", it's already formatted for SSE |
| 436 | if (data_str.substr(0, 6) == "data: ") { |
| 437 | chunk_content = data_str; |
| 438 | } else { |
| 439 | // Regular JSON string, add single newline for Ollama format |
| 440 | chunk_content = data_str; |
| 441 | } |
| 442 | } else { |
| 443 | // Regular JSON object, format for Ollama compatibility |
| 444 | chunk_content = data.dump() + "\n"; |
| 445 | } |
| 446 | |
| 447 | // std::cout << "Chunk: " << chunk_content; |
| 448 | |
| 449 | // HTTP chunked format: size in hex + \r\n + data + \r\n |
| 450 | std::ostringstream chunk_size; |
| 451 | chunk_size << std::hex << chunk_content.length(); |
| 452 | std::string http_chunk; |
| 453 | //if(is_final) |
| 454 | // http_chunk = chunk_content + "\r\n"; |
| 455 | //else |
| 456 | http_chunk = chunk_size.str() + "\r\n" + chunk_content + "\r\n"; |
| 457 | |
| 458 | |
| 459 | // Send chunk immediately |
| 460 | boost::system::error_code ec; |
| 461 | net::write(socket_, net::buffer(http_chunk), ec); |
| 462 | |
| 463 | if (ec) { |
| 464 | if (cancellation_token_ && !cancellation_token_->completed()) { |
| 465 | cancellation_token_->cancel(); |
| 466 | } |
| 467 | |
| 468 | //socket_.shutdown(tcp::socket::shutdown_both, ec); |
| 469 | //server_.active_connections_.fetch_sub(1); |
| 470 | |
| 471 | boost::system::error_code ignore_ec; |
| 472 | socket_.close(ignore_ec); |
| 473 | return; |
| 474 | } |
| 475 | |
| 476 | if (is_final) { |
| 477 | // Send final chunk (0-length chunk to end stream) |
| 478 | std::string final_chunk = "0\r\n\r\n"; |
| 479 | net::write(socket_, net::buffer(final_chunk), ec); |
| 480 | header_print("🔒 ", "Closing TCP connection (streaming, non-keep-alive)"); |
| 481 | socket_.shutdown(tcp::socket::shutdown_both, ec); |
| 482 | // Decrement connection counter for non-keep-alive connections |
| 483 | server_.active_connections_.fetch_sub(1); |
| 484 | } |
| 485 | } |
| 486 |