server_common/wsproxy
5
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

1

Browse Source
This commit is contained in:
aozhiwei 2023-04-24 11:07:22 +08:00
parent 54336693e4
commit 5ed753636c
4 changed files with 14 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
server/wsproxy/app.cc
View File

@ -420,8 +420,8 @@ void App::ProcessClientMsg(f8::MsgHdr& hdr)
} else {
auto down_wp = DownStreamMgr::Instance()->GetDownStream(hdr.socket_handle);
if (auto down = down_wp.lock(); !down_wp.expired()) {
if (!down->conn.expired()) {
down->conn.lock()->ForwardClientMsg(hdr);
if (!down->GetUpStream().expired()) {
down->GetUpStream().lock()->ForwardClientMsg(hdr);
}
}
}

6
server/wsproxy/downstream.cc
View File

@ -12,6 +12,8 @@ void DownStream::ForwardUpStreamMsg(f8::MsgHdr& hdr)
void DownStream::OnClose()
{
ss::SS_WSP_SocketDisconnect msg;
conn.lock()->SendMsg(socket_handle, msg);
if (!GetUpStream().expired()) {
ss::SS_WSP_SocketDisconnect msg;
GetUpStream().lock()->SendMsg(socket_handle, msg);
}
}

4
server/wsproxy/downstream.h
View File

@ -5,11 +5,13 @@ class DownStream
{
public:
int socket_handle = a8::INVALID_SOCKET_HANDLE;
std::weak_ptr<UpStream> conn;
void SetUpStream(std::weak_ptr<UpStream> up) { up_ = up; }
std::weak_ptr<UpStream> GetUpStream() { return up_; }
void ForwardUpStreamMsg(f8::MsgHdr& hdr);
void OnClose();
private:
int type_ = 0;
std::weak_ptr<UpStream> up_;
};

11
server/wsproxy/downstreammgr.cc
View File

@ -43,9 +43,7 @@ void DownStreamMgr::OnClientDisconnect(int socket_handle)
{
auto down_wp = GetDownStream(socket_handle);
if (auto down = down_wp.lock(); !down_wp.expired()) {
if (!down->conn.expired()) {
down->OnClose();
}
down->OnClose();
socket_hash_.erase(socket_handle);
}
RemovePendingAccount(socket_handle);
@ -56,7 +54,8 @@ void DownStreamMgr::OnUpStreamDisconnect(int instance_id)
{
std::list<std::shared_ptr<DownStream>> delete_client;
for (auto& pair : socket_hash_) {
if (!pair.second->conn.expired() && pair.second->conn.lock()->instance_id == instance_id) {
if (!pair.second->GetUpStream().expired() &&
pair.second->GetUpStream().lock()->instance_id == instance_id) {
delete_client.push_back(pair.second);
}
}
@ -84,11 +83,11 @@ void DownStreamMgr::BindUpStream(int socket_handle, int conn_instance_id)
if (!conn.expired()) {
auto down_wp = GetDownStream(socket_handle);
if (auto down = down_wp.lock(); !down_wp.expired()) {
down->conn = conn;
down->SetUpStream(conn);
} else {
down = std::make_shared<DownStream>();
down->socket_handle = socket_handle;
down->conn = conn;
down->SetUpStream(conn);
socket_hash_[down->socket_handle] = down;
f8::UdpLog::Instance()->Info("BindUpStream socket_handle:%d",
{