IM项目------消息存储子服务
提示：文章写完后，目录可以自动生成，如何生成可参考右边的帮助文档
文章目录
- 前言
- 消息存储表
- 消息存储表操作句柄
- ES消息操作句柄
- 创建索引
- 添加数据
- 删除数据
- 查询消息
- Message服务类
- rpc业务代码
- 获取指定会话的最近N条消息
- 获取用户信息
- 获取文件内容
- 获取指定会话的时间返回消息
- 关键字消息搜索
前言
消息存储子服务通过MQ消费者客户端订阅队列消息，来消费消息转发子服务发来的消息。需要将消息存储进mysql数据库/ES搜索引擎，另外文件消息的文件需要通过文件子服务存储进行上传。
消息存储子服务主要负责消息的存储以及提供消息的各种操作。提供三个服务：获取指定会话的最近N条消息/获取指定会话时间范围内的消息/按关键字搜索消息。
消息存储表
包含的字段：消息ID,会话ID,发送者ID,消息类型，产生时间，消息内容，文件ID,文件名称，文件大小。
其中会话ID需要添加索引，后面查询消息都是按会话ID来查询的。
消息内容是只有文本消息才填写的。
文件ID是图片/语音/文件消息才有的，当收到这个类型的消息，需要上传到文件存贮子服务上去，同时返回文件ID
文件名称和文件大小都是文件消息才有的字段。
#pragma db id auto
unsigned long _id;
#pragma db type("varchar(64)") index unique
std::string _message_id;
#pragma db type("varchar(64)") index
std::string _session_id; //所属会话ID
#pragma db type("varchar(64)")
std::string _user_id; //发送者用户ID
unsigned char _message_type; //消息类型 0-文本；1-图片；2-文件；3-语音
#pragma db type("TIMESTAMP")
boost::posix_time::ptime _create_time; //消息的产生时间odb::nullable<std::string> _content; //文本消息内容--非文本消息可以忽略
#pragma db type("varchar(64)")
odb::nullable<std::string> _file_id; //文件消息的文件ID -- 文本消息忽略
#pragma db type("varchar(128)")
odb::nullable<std::string> _file_name; //文件消息的文件名称 -- 只针对文件消息有效
odb::nullable<unsigned int> _file_size; //文件消息的文件大小 -- 只针对文件消息有效
消息存储表操作句柄
需要提供四个操作：
//• 新增消息 ---- 从MQ中消费一条消息时执行
//• 删除指定会话消息 ---- 删除好友/群聊时执行
//• 通过会话 ID，时间范围，获取指定时间段之内的消息，并按时间进行排序
//• 通过会话 ID，消息数量，获取最近的 N 条消息（逆序+limit 即可)
新增消息，外部直接传入一个Message对象进行插入。
bool insert(Message &msg) {try {odb::transaction trans(_db->begin());_db->persist(msg);trans.commit();}catch (std::exception &e) {LOG_ERROR("新增消息失败 {}:{}！", msg.message_id(),e.what());return false;}return true;
}
删除消息，通过会话ID进行删除。当删除好友或者删除群聊时会话被删除了，就需要删除指定会话的消息。
bool remove(const std::string &ssid) {try {odb::transaction trans(_db->begin());typedef odb::query<Message> query;typedef odb::result<Message> result;_db->erase_query<Message>(query::session_id == ssid);trans.commit();}catch (std::exception &e) {LOG_ERROR("删除会话所有消息失败 {}:{}！", ssid, e.what());return false;}return true;
}
获取指定会话的最近N条消息。根据消息产生时间降序，并且Limit指定个数，就可以获取最近N条消息。
返回一个Message数组，需要给这个数组逆序一下哎，方便前端进行渲染。
std::vector<Message> recent(const std::string &ssid, int count) {std::vector<Message> res;try {odb::transaction trans(_db->begin());typedef odb::query<Message> query;typedef odb::result<Message> result;//本次查询是以ssid作为过滤条件，然后进行以时间字段进行逆序，通过limit// session_id='xx' order by create_time desc limit count;std::stringstream cond;cond << "session_id='" << ssid << "' ";cond << "order by create_time desc limit " << count;result r(_db->query<Message>(cond.str()));for (result::iterator i(r.begin()); i != r.end(); ++i) {res.push_back(*i);}std::reverse(res.begin(), res.end());trans.commit();}catch (std::exception &e) {LOG_ERROR("获取最近消息失败:{}-{}-{}！", ssid, count, e.what());}return res;
}
获取指定会话的时间范围内的消息。就是where >=stime <=etime就行。
也是返回一个MEssage数组。
std::vector<Message> range(const std::string &ssid, boost::posix_time::ptime &stime, boost::posix_time::ptime &etime) {std::vector<Message> res;try {odb::transaction trans(_db->begin());typedef odb::query<Message> query;typedef odb::result<Message> result;//获取指定会话指定时间段的信息result r(_db->query<Message>(query::session_id == ssid && query::create_time >= stime &&query::create_time <= etime));for (result::iterator i(r.begin()); i != r.end(); ++i) {res.push_back(*i);}trans.commit();}catch (std::exception &e) {//将ptime类型转换为string类型进行日志打印LOG_ERROR("获取区间消息失败:{}-{}:{}-{}！", ssid, boost::posix_time::to_simple_string(stime), boost::posix_time::to_simple_string(etime), e.what());}return res;
}
ES消息操作句柄
需要把文本消息存储进ES搜索引擎中，方便前端进行关键字的消息查询，放进ES中是由于通过MYSQL的模糊查询效率太低了。
前面我们以及封装了一个ES操作类了，我们需要对这个类进一步封装，让他更加贴合与我们消息存储子服务。
创建索引
创建索引，索引字段需要有用户ID,消息ID,产生时间，会话ID,消息内容。
正文的是需要分词的，且需要参与索引。会话ID也是需要参与索引的。前端是查询指定会话的指定关键字的消息。
这里存入用户ID/消息ID/产生时间/是为了方便我们查询到消息后直接构建完整的消息类型。通过用户ID向用户子服务获取用户消息就可以构建完整的消息类型了。
bool createIndex() {bool ret = ESIndex(_es_client, "message").append("user_id", "keyword", "standard", false).append("message_id", "keyword", "standard", false).append("create_time", "long", "standard", false).append("chat_session_id", "keyword", "standard", true).append("content").create();if (ret == false) {LOG_INFO("消息信息索引创建失败!");return false;}LOG_INFO("消息信息索引创建成功!");return true;
}
添加数据
从消息队列获取到消息后，如果是文本消息则插入进ES中。这里是通过消息ID作为文档ID进行插入的。
bool appendData(const std::string &user_id,const std::string &message_id,const long create_time,const std::string &chat_session_id,const std::string &content) {bool ret = ESInsert(_es_client, "message").append("message_id", message_id).append("create_time", create_time).append("user_id", user_id).append("chat_session_id", chat_session_id).append("content", content).insert(message_id);if (ret == false) {LOG_ERROR("消息数据插入/更新失败!");return false;}LOG_INFO("消息数据新增/更新成功!");return true;
}
删除数据
通过文档ID也就是消息ID进行删除。
bool remove(const std::string &mid) {bool ret = ESRemove(_es_client, "message").remove(mid);if (ret == false) {LOG_ERROR("消息数据删除失败!");return false;}LOG_INFO("消息数据删除成功!");return true;
}
查询消息
通过前端输入的关键字和会话ID进行查询，因为在ES中存储的消息产生时间是一个时间戳，在我们odb映射类型中是ptime，需要进行一个转换。返回值是一个Message数组。这里ES查询数据和我们前面查询用户有些不同。
查询用户需要设置一个must_not和should参数。
这里查询消息，是通过must_term和must_match字段进行查询的。
must_term是精确匹配，一般用于keyword字段，我们这是是会话ID字段必须精确匹配。
must_match是全文搜索，一般用于text字段。也就是content字段必须匹配用户的关键字。
std::vector<lkm_im::Message> search(const std::string &key, const std::string &ssid) {std::vector<lkm_im::Message> res;Json::Value json_user = ESSearch(_es_client, "message").append_must_term("chat_session_id.keyword", ssid).append_must_match("content", key).search();if (json_user.isArray() == false) {LOG_ERROR("用户搜索结果为空，或者结果不是数组类型");return res;}int sz = json_user.size();LOG_DEBUG("检索结果条目数量：{}", sz);for (int i = 0; i < sz; i++) {lkm_im::Message message;message.user_id(json_user[i]["_source"]["user_id"].asString());message.message_id(json_user[i]["_source"]["message_id"].asString());boost::posix_time::ptime ctime(boost::posix_time::from_time_t(json_user[i]["_source"]["create_time"].asInt64()));message.create_time(ctime);message.session_id(json_user[i]["_source"]["chat_session_id"].asString());message.content(json_user[i]["_source"]["content"].asString());res.push_back(message);}return res;
}
Message服务类
消息存贮子服务不光要搭建一个rpc服务器。还有一个MQ消费者客户端。
这个客户端订阅了消息转发子服务创建的队列，当消息转发子服务收到一个新消息后，就会把该消息生产进消息队列中。
所以我们的服务类中，需要订阅这个队列，同时提供一个回调函数，当消息来临进行消费。
根据消息首地址和消息长度提取出完整的报文，进行反序列化，提取出完整的消息类型。
根据不同的消息类型进行不同的处理，
如果是文本消息，则需要存储进ES中。同时需要存储进MYsql消息表中。这里文本消息所需要的字段都有，直接存进mysql就行。
如果是其他类型的消息，在Mysql中是需要存一个文件ID的，我们需要把文件内容上传到文件存储子服务中，获取返回的文件ID，在Mysql进行一个存储。另外如果是文件类型的消息，还需要存入文件名和文件大小。
这里提一下上传消息的req中有一个FileUploadData类型，这个类型就是需要提供文件名称，文件大小，文件内容。而我们的图片类型消息和语音类型消息是没有文件名称和文件大小的。所以这里我们在上传文件时填入空。
proto中定义的消息类型中消息创建时间是一个时间戳int64类型，在mysql映射类时ptime因此需要进行转换。
void onMessage(const char *body, size_t sz) {LOG_DEBUG("收到新消息，进行存储处理！");//1. 取出序列化的消息内容，进行反序列化lkm_im::MessageInfo message;bool ret = message.ParseFromArray(body, sz);if (ret == false) {LOG_ERROR("对消费到的消息进行反序列化失败！");return;}//2. 根据不同的消息类型进行不同的处理std::string file_id, file_name, content;int64_t file_size;switch(message.message().message_type()) {// 1. 如果是一个文本类型消息，取元信息存储到ES中case MessageType::STRING:content = message.message().string_message().content();ret = _es_message->appendData(message.sender().user_id(),message.message_id(),message.timestamp(),message.chat_session_id(),content);if (ret == false) {LOG_ERROR("文本消息向存储引擎进行存储失败！");return;}break;// 2. 如果是一个图片/语音/文件消息，则取出数据存储到文件子服务中，并获取文件IDcase MessageType::IMAGE:{const auto &msg = message.message().image_message();ret = _PutFile("", msg.image_content(), msg.image_content().size(), file_id);if (ret == false) {LOG_ERROR("上传图片到文件子服务失败！");return ;}}break;case MessageType::FILE:{const auto &msg = message.message().file_message();file_name = msg.file_name();file_size = msg.file_size();ret = _PutFile(file_name, msg.file_contents(), file_size, file_id);if (ret == false) {LOG_ERROR("上传文件到文件子服务失败！");return ;}}break;case MessageType::SPEECH:{const auto &msg = message.message().speech_message();ret = _PutFile("", msg.file_contents(), msg.file_contents().size(), file_id);if (ret == false) {LOG_ERROR("上传语音到文件子服务失败！");return ;}}break;default:LOG_ERROR("消息类型错误！");return;}//3. 提取消息的元信息，存储到mysql数据库中lkm_im::Message msg(message.message_id(), message.chat_session_id(),message.sender().user_id(),message.message().message_type(),boost::posix_time::from_time_t(message.timestamp()));msg.content(content);msg.file_id(file_id);msg.file_name(file_name);msg.file_size(file_size);ret = _mysql_message->insert(msg);if (ret == false) {LOG_ERROR("向数据库插入新消息失败！");return;}}
rpc业务代码
消息存储自服务提供了三个服务：
- 获取指定会话的最近N条消息
- 获取指定会话的时间范围的消息
- 通过关键字查询指定会话的消息
获取指定会话的最近N条消息
先从req中提取出会话ID和获取消息个数。
去Mysql消息表查询，获取到一个Message数组。
因为我们要组织完整的消息结构，所以需要获取到发送者信息。另外如果时文件消息，还需要从文件存储子服务获取文件。
获取用户信息
这里需要提供请求ID,和一个用户ID列表。请求ID就是客户端发来的请求Id,用户列表就是mysql查出的Message数组里的用户ID。
我们调用获取一组用户信息的服务，响应中会返回一个 goole::protobuf::map<string, UserInfo> 数组。我们遍历这个数组，把他插入到我们自己的std::unordered_map中，其中first就是用户ID,second就是用户信息。
bool _GetUser(const std::string &rid,const std::unordered_set<std::string> &user_id_lists,std::unordered_map<std::string, UserInfo> &user_lists) {auto channel = _mm_channels->choose(_user_service_name);if (!channel) {LOG_ERROR("{} 没有可供访问的用户子服务节点！", _user_service_name);return false;}UserService_Stub stub(channel.get());GetMultiUserInfoReq req;GetMultiUserInfoRsp rsp;req.set_request_id(rid);for (const auto &id : user_id_lists) {req.add_users_id(id);}brpc::Controller cntl;stub.GetMultiUserInfo(&cntl, &req, &rsp, nullptr);if (cntl.Failed() == true || rsp.success() == false) {LOG_ERROR("用户子服务调用失败：{}！", cntl.ErrorText());return false;}const auto &umap = rsp.users_info();for (auto it = umap.begin(); it != umap.end(); ++it) {user_lists.insert(std::make_pair(it->first, it->second));}return true;
}
获取文件内容
获取文件和获取用户一样的流程。也是提供一组用户ID,这里rsp返回的是map<string, FileDownloadData>，我们遍历这个map提取出FileDownloadData中的文件内容。插入到我们的std::unordered_map中，其中first是文件ID,second是content文件内容。
bool _GetFile(const std::string &rid,const std::unordered_set<std::string> &file_id_lists,std::unordered_map<std::string, std::string> &file_data_lists) {auto channel = _mm_channels->choose(_file_service_name);if (!channel) {LOG_ERROR("{} 没有可供访问的文件子服务节点！", _file_service_name);return false;}FileService_Stub stub(channel.get());GetMultiFileReq req;GetMultiFileRsp rsp;req.set_request_id(rid);for (const auto &id : file_id_lists) {req.add_file_id_list(id);}brpc::Controller cntl;stub.GetMultiFile(&cntl, &req, &rsp, nullptr);if (cntl.Failed() == true || rsp.success() == false) {LOG_ERROR("文件子服务调用失败：{}！", cntl.ErrorText());return false;}const auto &fmap = rsp.file_data();for (auto it = fmap.begin(); it != fmap.end(); ++it) {file_data_lists.insert(std::make_pair(it->first, it->second.file_content()));}return true;
}
在获取完用户信息Map和文件内容Map后我们就可以组织完整的消息结构数组进行返回了。
virtual void GetRecentMsg(::google::protobuf::RpcController* controller,const ::lkm_im::GetRecentMsgReq* request,::lkm_im::GetRecentMsgRsp* response,::google::protobuf::Closure* done) {brpc::ClosureGuard rpc_guard(done);auto err_response = [this, response](const std::string &rid, const std::string &errmsg) -> void {response->set_request_id(rid);response->set_success(false);response->set_errmsg(errmsg);return;};//1. 提取请求中的关键要素：请求ID，会话ID，要获取的消息数量std::string rid = request->request_id();std::string chat_ssid = request->chat_session_id();int msg_count = request->msg_count();//2. 从数据库，获取最近的消息元信息auto msg_lists = _mysql_message->recent(chat_ssid, msg_count);if (msg_lists.empty()) {response->set_request_id(rid);response->set_success(true);return ;}//3. 统计所有消息中文件类型消息的文件ID列表，从文件子服务下载文件std::unordered_set<std::string> file_id_lists;for (const auto &msg : msg_lists) {if (msg.file_id().empty()) continue;LOG_DEBUG("需要下载的文件ID: {}", msg.file_id());file_id_lists.insert(msg.file_id());}std::unordered_map<std::string, std::string> file_data_lists;bool ret = _GetFile(rid, file_id_lists, file_data_lists);if (ret == false) {LOG_ERROR("{} 批量文件数据下载失败！", rid);return err_response(rid, "批量文件数据下载失败!");}//4. 统计所有消息的发送者用户ID，从用户子服务进行批量用户信息获取std::unordered_set<std::string> user_id_lists;for (const auto &msg : msg_lists) {user_id_lists.insert(msg.user_id());}std::unordered_map<std::string, UserInfo> user_lists;ret = _GetUser(rid, user_id_lists, user_lists);if (ret == false) {LOG_ERROR("{} 批量用户数据获取失败！", rid);return err_response(rid, "批量用户数据获取失败!");}//5. 组织响应response->set_request_id(rid);response->set_success(true);for (const auto &msg : msg_lists) {auto message_info = response->add_msg_list();message_info->set_message_id(msg.message_id());message_info->set_chat_session_id(msg.session_id());message_info->set_timestamp(boost::posix_time::to_time_t(msg.create_time()));message_info->mutable_sender()->CopyFrom(user_lists[msg.user_id()]);switch(msg.message_type()) {case MessageType::STRING:message_info->mutable_message()->set_message_type(MessageType::STRING);message_info->mutable_message()->mutable_string_message()->set_content(msg.content());break;case MessageType::IMAGE:message_info->mutable_message()->set_message_type(MessageType::IMAGE);message_info->mutable_message()->mutable_image_message()->set_file_id(msg.file_id());message_info->mutable_message()->mutable_image_message()->set_image_content(file_data_lists[msg.file_id()]);break;case MessageType::FILE:message_info->mutable_message()->set_message_type(MessageType::FILE);message_info->mutable_message()->mutable_file_message()->set_file_id(msg.file_id());message_info->mutable_message()->mutable_file_message()->set_file_size(msg.file_size());message_info->mutable_message()->mutable_file_message()->set_file_name(msg.file_name());message_info->mutable_message()->mutable_file_message()->set_file_contents(file_data_lists[msg.file_id()]);break;case MessageType::SPEECH:message_info->mutable_message()->set_message_type(MessageType::SPEECH);message_info->mutable_message()->mutable_speech_message()->set_file_id(msg.file_id());message_info->mutable_message()->mutable_speech_message()->set_file_contents(file_data_lists[msg.file_id()]);break;default:LOG_ERROR("消息类型错误！！");return;}}return;
}
获取指定会话的时间返回消息
这里和获取指定会话的最近N条消息一致的思想。
也是获取到会话ID和时间返回。
通过MYsql查询到一个MEssage数组，
由于要组织完整的消息结构，需要从用户子服务获取到用户信息，从文件子服务获取到文件内容。
然后组织消息进行返回
virtual void GetHistoryMsg(::google::protobuf::RpcController* controller,const ::lkm_im::GetHistoryMsgReq* request,::lkm_im::GetHistoryMsgRsp* response,::google::protobuf::Closure* done) {brpc::ClosureGuard rpc_guard(done);auto err_response = [this, response](const std::string &rid, const std::string &errmsg) -> void {response->set_request_id(rid);response->set_success(false);response->set_errmsg(errmsg);return;};//1. 提取关键要素：会话ID，起始时间，结束时间std::string rid = request->request_id();std::string chat_ssid = request->chat_session_id();boost::posix_time::ptime stime = boost::posix_time::from_time_t(request->start_time());boost::posix_time::ptime etime = boost::posix_time::from_time_t(request->over_time());//2. 从数据库中进行消息查询auto msg_lists = _mysql_message->range(chat_ssid, stime, etime);if (msg_lists.empty()) {response->set_request_id(rid);response->set_success(true);return ;}//3. 统计所有文件类型消息的文件ID，并从文件子服务进行批量文件下载std::unordered_set<std::string> file_id_lists;for (const auto &msg : msg_lists) {if (msg.file_id().empty()) continue;LOG_DEBUG("需要下载的文件ID： {}", msg.file_id());file_id_lists.insert(msg.file_id());}std::unordered_map<std::string, std::string> file_data_lists;bool ret = _GetFile(rid, file_id_lists, file_data_lists);if (ret == false) {LOG_ERROR("{} 批量文件数据下载失败！", rid);return err_response(rid, "批量文件数据下载失败!");}//4. 统计所有消息的发送者用户ID，从用户子服务进行批量用户信息获取std::unordered_set<std::string> user_id_lists; // {猪爸爸吧， 祝妈妈，猪爸爸吧，祝爸爸}for (const auto &msg : msg_lists) {user_id_lists.insert(msg.user_id());}std::unordered_map<std::string, UserInfo> user_lists;ret = _GetUser(rid, user_id_lists, user_lists);if (ret == false) {LOG_ERROR("{} 批量用户数据获取失败！", rid);return err_response(rid, "批量用户数据获取失败!");}//5. 组织响应response->set_request_id(rid);response->set_success(true);for (const auto &msg : msg_lists) {auto message_info = response->add_msg_list();message_info->set_message_id(msg.message_id());message_info->set_chat_session_id(msg.session_id());message_info->set_timestamp(boost::posix_time::to_time_t(msg.create_time()));message_info->mutable_sender()->CopyFrom(user_lists[msg.user_id()]);switch(msg.message_type()) {case MessageType::STRING:message_info->mutable_message()->set_message_type(MessageType::STRING);message_info->mutable_message()->mutable_string_message()->set_content(msg.content());break;case MessageType::IMAGE:message_info->mutable_message()->set_message_type(MessageType::IMAGE);message_info->mutable_message()->mutable_image_message()->set_file_id(msg.file_id());message_info->mutable_message()->mutable_image_message()->set_image_content(file_data_lists[msg.file_id()]);break;case MessageType::FILE:message_info->mutable_message()->set_message_type(MessageType::FILE);message_info->mutable_message()->mutable_file_message()->set_file_id(msg.file_id());message_info->mutable_message()->mutable_file_message()->set_file_size(msg.file_size());message_info->mutable_message()->mutable_file_message()->set_file_name(msg.file_name());message_info->mutable_message()->mutable_file_message()->set_file_contents(file_data_lists[msg.file_id()]);break;case MessageType::SPEECH:message_info->mutable_message()->set_message_type(MessageType::SPEECH);message_info->mutable_message()->mutable_speech_message()->set_file_id(msg.file_id());message_info->mutable_message()->mutable_speech_message()->set_file_contents(file_data_lists[msg.file_id()]);break;default:LOG_ERROR("消息类型错误！！");return;}}return;
}
关键字消息搜索
这个功能相比与前两个更加简单了，因为关键字消息搜索只能搜索文本消息，所以免去了向文件子服务获取问价内容的过程。
首先提取出会话ID,和关键字。
在ES中进行查询，会返回一个Message数组。
根据数组中的用户ID查询用户子服务获取用户信息，然后组织响应返回。
virtual void MsgSearch(::google::protobuf::RpcController* controller,const ::lkm_im::MsgSearchReq* request,::lkm_im::MsgSearchRsp* response,::google::protobuf::Closure* done) {brpc::ClosureGuard rpc_guard(done);auto err_response = [this, response](const std::string &rid, const std::string &errmsg) -> void {response->set_request_id(rid);response->set_success(false);response->set_errmsg(errmsg);return;};//关键字的消息搜索--只针对文本消息//1. 从请求中提取关键要素：请求ID，会话ID, 关键字std::string rid = request->request_id();std::string chat_ssid = request->chat_session_id();std::string skey = request->search_key();//2. 从ES搜索引擎中进行关键字消息搜索，得到消息列表auto msg_lists = _es_message->search(skey, chat_ssid);if (msg_lists.empty()) {response->set_request_id(rid);response->set_success(true);return ;}//3. 组织所有消息的用户ID，从用户子服务获取用户信息std::unordered_set<std::string> user_id_lists;for (const auto &msg : msg_lists) {user_id_lists.insert(msg.user_id());}std::unordered_map<std::string, UserInfo> user_lists;bool ret = _GetUser(rid, user_id_lists, user_lists);if (ret == false) {LOG_ERROR("{} 批量用户数据获取失败！", rid);return err_response(rid, "批量用户数据获取失败!");}//4. 组织响应 response->set_request_id(rid);response->set_success(true);for (const auto &msg : msg_lists) {auto message_info = response->add_msg_list();message_info->set_message_id(msg.message_id());message_info->set_chat_session_id(msg.session_id());message_info->set_timestamp(boost::posix_time::to_time_t(msg.create_time()));message_info->mutable_sender()->CopyFrom(user_lists[msg.user_id()]);message_info->mutable_message()->set_message_type(MessageType::STRING);message_info->mutable_message()->mutable_string_message()->set_content(msg.content());}return;
}