Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Add magic #90

Merged
merged 2 commits into from
Dec 23, 2022
Merged

Add magic #90

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
e1662e3
add magic
Dec 23, 2022
dc8d790
check magic
Dec 23, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Next Next commit
add magic
  • Loading branch information
chufeng.qy@alibaba-inc.com committed Dec 23, 2022
commit e1662e3a9fc289e333424d17eab596abf4acbbd6
11 changes: 6 additions & 5 deletions include/rest_rpc/connection.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -228,11 +228,10 @@ class connection : public std::enable_shared_from_this<connection>,
void write() {
auto &msg = write_queue_.front();
write_size_ = (uint32_t)msg.content->size();
std::array<asio::const_buffer, 4> write_buffers;
write_buffers[0] = asio::buffer(&write_size_, sizeof(uint32_t));
write_buffers[1] = asio::buffer(&msg.req_id, sizeof(uint64_t));
write_buffers[2] = asio::buffer(&msg.req_type, sizeof(request_type));
write_buffers[3] = asio::buffer(msg.content->data(), write_size_);
header_ = {MAGIC_NUM, msg.req_type, write_size_, msg.req_id};
std::array<asio::const_buffer, 2> write_buffers;
write_buffers[0] = asio::buffer(&header_, sizeof(rpc_header));
write_buffers[1] = asio::buffer(msg.content->data(), write_size_);

auto self = this->shared_from_this();
async_write(write_buffers,
Expand Down Expand Up @@ -397,6 +396,8 @@ class connection : public std::enable_shared_from_this<connection>,
std::uint64_t req_id_;
request_type req_type_;

rpc_header header_;

uint32_t write_size_ = 0;
std::mutex write_mtx_;

Expand Down
8 changes: 4 additions & 4 deletions include/rest_rpc/const_vars.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,15 +25,15 @@ struct message_type {
std::shared_ptr<std::string> content;
};

#pragma pack(1)
static const uint8_t MAGIC_NUM = 39;
struct rpc_header {
uint8_t magic;
request_type req_type;
uint32_t body_len;
uint64_t req_id;
request_type req_type;
};
#pragma pack()

static const size_t MAX_BUF_LEN = 1048576 * 10;
static const size_t HEAD_LEN = 13;
static const size_t HEAD_LEN = sizeof(rpc_header);
static const size_t INIT_BUF_SIZE = 2 * 1024;
} // namespace rest_rpc
17 changes: 9 additions & 8 deletions include/rest_rpc/rpc_client.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -445,11 +445,10 @@ class rpc_client : private asio::noncopyable {
void write() {
auto &msg = outbox_[0];
write_size_ = (uint32_t)msg.content.length();
std::array<asio::const_buffer, 4> write_buffers;
write_buffers[0] = asio::buffer(&write_size_, sizeof(int32_t));
write_buffers[1] = asio::buffer(&msg.req_id, sizeof(uint64_t));
write_buffers[2] = asio::buffer(&msg.req_type, sizeof(request_type));
write_buffers[3] = asio::buffer((char *)msg.content.data(), write_size_);
std::array<asio::const_buffer, 2> write_buffers;
header_ = {MAGIC_NUM, msg.req_type, write_size_, msg.req_id};
write_buffers[0] = asio::buffer(&header_, sizeof(rpc_header));
write_buffers[1] = asio::buffer((char *)msg.content.data(), write_size_);

async_write(write_buffers,
[this](const asio::error_code &ec, const size_t length) {
Expand Down Expand Up @@ -508,14 +507,14 @@ class rpc_client : private asio::noncopyable {
}
} else {
std::cout << ec.message() << "\n";

{
std::unique_lock<std::mutex> lock(cb_mtx_);
for (auto& item : callback_map_) {
for (auto &item : callback_map_) {
item.second->callback(ec, {});
}
}

close(false);
error_callback(ec);
}
Expand Down Expand Up @@ -854,6 +853,8 @@ class rpc_client : private asio::noncopyable {
char head_[HEAD_LEN] = {};
std::vector<char> body_;

rpc_header header_;

std::unordered_map<std::string, std::function<void(string_view)>> sub_map_;
std::set<std::pair<std::string, std::string>> key_token_set_;

Expand Down