/**
|
* Copyright FunASR (https://github.com/alibaba-damo-academy/FunASR). All Rights
|
* Reserved. MIT License (https://opensource.org/licenses/MIT)
|
*/
|
/* 2023-2025 by zhaomingwork@qq.com */
|
//
|
// connection.cpp
|
// copy some codes from http://www.boost.org/
|
#include "connection.hpp"
|
|
#include <thread>
|
#include <utility>
|
#include "util.hpp"
|
|
|
namespace http
|
{
|
namespace server2
|
{
|
|
std::shared_ptr<FUNASR_MESSAGE> &connection::get_data_msg() { return data_msg; }
|
connection::connection(asio::ip::tcp::socket socket,
|
asio::io_context &io_decoder, int connection_id,
|
std::shared_ptr<ModelDecoder> model_decoder)
|
: socket_(std::move(socket)),
|
io_decoder(io_decoder),
|
connection_id(connection_id),
|
model_decoder(model_decoder)
|
|
{
|
s_timer = std::make_shared<asio::steady_timer>(io_decoder);
|
}
|
|
void connection::setup_timer()
|
{
|
if (data_msg->status == 1)
|
return;
|
|
s_timer->expires_after(std::chrono::seconds(10));
|
s_timer->async_wait([=](const asio::error_code &ec)
|
{
|
if (!ec) {
|
std::cout << "time is out!" << std::endl;
|
if (data_msg->status == 1) return;
|
data_msg->status = 1;
|
s_timer->cancel();
|
auto wf = std::bind(&connection::write_back, std::ref(*this), "");
|
// close the connection
|
strand_->post(wf);
|
} });
|
}
|
|
void connection::start()
|
{
|
std::lock_guard<std::mutex> lock(m_lock); // for threads safty
|
try
|
{
|
|
data_msg = std::make_shared<FUNASR_MESSAGE>(); // put a new data vector for
|
// new connection
|
data_msg->samples = std::make_shared<std::vector<char>>();
|
// data_msg->samples->reserve(16000*20);
|
data_msg->msg = nlohmann::json::parse("{}");
|
data_msg->msg["wav_format"] = "pcm";
|
data_msg->msg["wav_name"] = "wav-default-id";
|
data_msg->msg["itn"] = true;
|
data_msg->msg["audio_fs"] = 16000; // default is 16k
|
data_msg->msg["access_num"] = 0; // the number of access for this object,
|
// when it is 0, we can free it saftly
|
data_msg->msg["is_eof"] = false;
|
data_msg->status = 0;
|
|
strand_ = std::make_shared<asio::io_context::strand>(io_decoder);
|
|
FUNASR_DEC_HANDLE decoder_handle = FunASRWfstDecoderInit(
|
model_decoder->get_asr_handle(), ASR_OFFLINE, global_beam_, lattice_beam_, am_scale_);
|
|
data_msg->decoder_handle = decoder_handle;
|
|
if (data_msg->hotwords_embedding == nullptr)
|
{
|
std::unordered_map<std::string, int> merged_hws_map;
|
std::string nn_hotwords = "";
|
|
if (true)
|
{
|
std::string json_string = "{}";
|
if (!json_string.empty())
|
{
|
nlohmann::json json_fst_hws;
|
try
|
{
|
json_fst_hws = nlohmann::json::parse(json_string);
|
if (json_fst_hws.type() == nlohmann::json::value_t::object)
|
{
|
// fst
|
try
|
{
|
std::unordered_map<std::string, int> client_hws_map =
|
json_fst_hws;
|
merged_hws_map.insert(client_hws_map.begin(),
|
client_hws_map.end());
|
}
|
catch (const std::exception &e)
|
{
|
std::cout << e.what();
|
}
|
}
|
}
|
catch (std::exception const &e)
|
{
|
std::cout << e.what();
|
// nn
|
std::string client_nn_hws = "{}";
|
nn_hotwords += " " + client_nn_hws;
|
std::cout << "nn hotwords: " << client_nn_hws;
|
}
|
}
|
}
|
merged_hws_map.insert(hws_map_.begin(), hws_map_.end());
|
|
// fst
|
std::cout << "hotwords: ";
|
for (const auto &pair : merged_hws_map)
|
{
|
nn_hotwords += " " + pair.first;
|
std::cout << pair.first << " : " << pair.second;
|
}
|
FunWfstDecoderLoadHwsRes(data_msg->decoder_handle, fst_inc_wts_,
|
merged_hws_map);
|
|
// nn
|
std::vector<std::vector<float>> new_hotwords_embedding =
|
CompileHotwordEmbedding(model_decoder->get_asr_handle(), nn_hotwords);
|
data_msg->hotwords_embedding =
|
std::make_shared<std::vector<std::vector<float>>>(
|
new_hotwords_embedding);
|
}
|
|
|
do_read();
|
}
|
catch (const std::exception &e)
|
{
|
std::cout << "error:" << e.what();
|
}
|
}
|
|
void connection::write_back(std::string str)
|
{
|
s_timer->cancel();
|
|
reply_ = reply::stock_reply(
|
data_msg->msg["asr_result"].dump()); // reply::stock_reply();
|
do_write();
|
}
|
void connection::do_read()
|
{
|
|
if (data_msg->status == 1)
|
return;
|
|
s_timer->cancel();
|
setup_timer();
|
auto self(shared_from_this());
|
socket_.async_read_some(
|
asio::buffer(buffer_),
|
[this, self](asio::error_code ec, std::size_t bytes_transferred)
|
{
|
if (ec)
|
{
|
handle_error(ec);
|
return;
|
}
|
|
// 将新数据追加到累积缓冲区
|
received_data_.append(buffer_.data(), bytes_transferred);
|
|
switch (state_)
|
{
|
case State::ReadingHeaders:
|
|
if (try_parse_headers())
|
{
|
if (state_ == State::SendingContinue)
|
{
|
|
handle_100_continue();
|
}
|
else
|
{
|
|
handle_body();
|
}
|
}
|
else
|
{
|
|
do_read();
|
}
|
break;
|
|
case State::ReadingBody:
|
|
handle_body();
|
break;
|
|
case State::SendingContinue:
|
|
break; // 等待100 Continue发送完成
|
}
|
});
|
}
|
|
std::string connection::parse_attachment_filename(const std::string &header)
|
{
|
|
size_t pos = header.find("Content-Disposition: ");
|
if (pos == std::string::npos)
|
return "";
|
|
pos += 21; // "Content-Disposition: "长度
|
size_t end = header.find("\r\n", pos);
|
if (end == std::string::npos)
|
return "";
|
|
// 调用解析函数
|
return parse_attachment_filename_impl(header.substr(pos, end - pos));
|
}
|
|
void connection::handle_body()
|
{
|
|
|
process_multipart_data();
|
if (in_file_part_ == false)
|
{
|
std::cout << "文件获取结束" << std::endl;
|
std::cout << "开始解码，数据大小= " << data_msg->samples->size() << std::endl;
|
|
auto close_thread = std::bind(&connection::write_back, std::ref(*this), "close");
|
auto decoder_thread = std::bind(&ModelDecoder::do_decoder,
|
std::ref(*model_decoder), std::ref(data_msg));
|
|
// for decode task
|
strand_->post(decoder_thread);
|
// for close task
|
strand_->post(close_thread);
|
data_msg->sem_resultok.acquire();
|
std::cout << "解码线程提交结束！！！！ " << std::endl;
|
}
|
else
|
do_read();
|
return;
|
|
}
|
|
// 辅助函数：解析 Content-Length
|
size_t connection::parse_content_length(const std::string &header)
|
{
|
size_t pos = header.find("Content-Length: ");
|
if (pos == std::string::npos)
|
return 0;
|
|
pos += 16; // "Content-Length: "长度
|
size_t end = header.find("\r\n", pos);
|
if (end == std::string::npos)
|
return 0;
|
|
try
|
{
|
return std::stoul(header.substr(pos, end - pos));
|
}
|
catch (...)
|
{
|
return 0;
|
}
|
}
|
|
void connection::handle_error(asio::error_code ec)
|
{
|
if (ec == asio::error::eof)
|
{
|
std::cout << "Connection closed gracefully\n";
|
}
|
else
|
{
|
std::cerr << "Error: " << ec.message() << "\n";
|
}
|
}
|
void connection::do_write()
|
{
|
auto self(shared_from_this());
|
asio::async_write(socket_, reply_.to_buffers(),
|
[this, self](asio::error_code ec, std::size_t)
|
{
|
if (!ec)
|
{
|
// Initiate graceful connection closure.
|
asio::error_code ignored_ec;
|
socket_.shutdown(asio::ip::tcp::socket::shutdown_both,
|
ignored_ec);
|
}
|
data_msg->sem_resultok.release();
|
// No new asynchronous operations are started. This means
|
// that all shared_ptr references to the connection object
|
// will disappear and the object will be destroyed
|
// automatically after this handler returns. The
|
// connection class's destructor closes the socket.
|
});
|
}
|
|
} // namespace server2
|
} // namespace http
|
