Currently, the SockUtil::connect function creates and binds the socket, then immediately connects to the remote destination IP and port. However, sometimes these two things need to be done separately. For example, during GB28181 playback negotiation, when the subordinate receives the invite request from the superior and returns a 200ok SDP message with the streaming source IP and port, some superiors have added security boundary devices that require verification of the subordinate's source IP and port before allowing the subordinate to connect and send stream data. In this case, three steps are required: 1. Create and bind the local port; 2. Send the streaming IP and port to the superior and wait for the superior's ack; 3. Connect to the remote destination IP and port. The specific implementation code is:
bool Socket::createAndBindSock(bool udpOrTcp, uint16_t port, const string &local_ip) {
closeSock();
int fd = SockUtil::createAndBindSock(udpOrTcp, port, local_ip.data());
if (fd == -1)
return false;
SockFD::Ptr sock;
if (udpOrTcp) {
sock = makeSock(fd, SockNum::Sock_UDP);
if (!attachEvent(sock, true))
return false;
}
else
{
// 注意tcp在连接目标成功后才会监听事件
sock = makeSock(fd, SockNum::Sock_TCP);
}
LOCK_GUARD(_mtx_sock_fd);
_sock_fd = sock;
return true;
}
void Socket::connect(const string &url, uint16_t port, onErrCB con_cb_in,
float timeout_sec, const string &local_ip, uint16_t local_port, bool newSock) {
if (newSock) {
// 重置当前socket
closeSock();
}
weak_ptr<Socket> weak_self = shared_from_this();
auto con_cb = [con_cb_in, weak_self](const SockException &err) {
auto strong_self = weak_self.lock();
if (!strong_self) {
return;
}
strong_self->_async_con_cb = nullptr;
strong_self->_con_timer = nullptr;
if (err) {
LOCK_GUARD(strong_self->_mtx_sock_fd);
strong_self->_sock_fd = nullptr;
}
con_cb_in(err);
};
auto async_con_cb = std::make_shared<function<void(int)> >([weak_self, con_cb, newSock](int sock) {
auto strong_self = weak_self.lock();
if (sock == -1 || !strong_self) {
if (!strong_self) {
CLOSE_SOCK(sock);
} else {
con_cb(SockException(Err_dns, get_uv_errmsg(true)));
}
return;
}
bool test = false;
SockFD::Ptr sock_fd = strong_self->getSock();
if (newSock || !sock_fd) {
test = true;
sock_fd = strong_self->makeSock(sock, SockNum::Sock_TCP);
}
weak_ptr<SockFD> weak_sock_fd = sock_fd;
// 监听该socket是否可写,可写表明已经连接服务器成功
int result = strong_self->_poller->addEvent(sock, Event_Write, [weak_self, weak_sock_fd, con_cb](int event) {
auto strong_sock_fd = weak_sock_fd.lock();
auto strong_self = weak_self.lock();
if (strong_sock_fd && strong_self) {
//socket可写事件,说明已经连接服务器成功
strong_self->onConnected(strong_sock_fd, con_cb);
}
});
if (result == -1) {
con_cb(SockException(Err_other, "add event to poller failed when start connect"));
return;
}
if (test) {
// 保存fd
strong_self->setSock(sock_fd);
}
});
auto poller = _poller;
weak_ptr<function<void(int)> > weak_task = async_con_cb;
int sockfd_in = -1;
if (!newSock) {
SockFD::Ptr sock_fd = getSock();
if (sock_fd)
sockfd_in = sock_fd->rawFd();
}
WorkThreadPool::Instance().getExecutor()->async([url, port, local_ip, local_port, sockfd_in, weak_task, poller]() {
// 阻塞式dns解析放在后台线程执行
int sock = SockUtil::connect(url.data(), port, true, local_ip.data(), local_port, sockfd_in);
poller->async([sock, weak_task]() {
auto strong_task = weak_task.lock();
if (strong_task) {
(*strong_task)(sock);
} else {
CLOSE_SOCK(sock);
}
});
});
//连接超时定时器
_con_timer = std::make_shared<Timer>(timeout_sec, [weak_self, con_cb]() {
con_cb(SockException(Err_timeout, uv_strerror(UV_ETIMEDOUT)));
return false;
}, _poller);
_async_con_cb = async_con_cb;
}
int SockUtil::createAndBindSock(bool udpOrTcp, const uint16_t port, const char* localIp, int af, bool bAsync, bool reusePort) {
int sockfd = -1;
if (udpOrTcp)
sockfd = (int)socket(af, SOCK_DGRAM, IPPROTO_UDP);
else
sockfd = (int)socket(af, SOCK_STREAM, IPPROTO_TCP);
if (sockfd == -1) {
WarnL << "创建套接字失败:" << get_uv_errmsg(true);
return -1;
}
setReuseable(sockfd, reusePort);
setNoSigpipe(sockfd);
setNoBlocked(sockfd, bAsync);
if (!udpOrTcp)
setNoDelay(sockfd);
setSendBuf(sockfd);
setRecvBuf(sockfd);
setCloseWait(sockfd);
setCloExec(sockfd);
if (bindSock(sockfd, localIp, port) == -1) {
close(sockfd);
return -1;
}
return sockfd;
}
int SockUtil::connect(const char *host, uint16_t port, bool bAsync, const char* localIp, uint16_t localPort, int sockfd_in) {
sockaddr addr;
if(!DnsCache::Instance().getDomainIP(host,addr)){
//dns解析失败
return -1;
}
//设置端口号
((sockaddr_in *)&addr)->sin_port = htons(port);
int sockfd = sockfd_in;
if (sockfd < 0)
sockfd = createAndBindSock(false, localPort, localIp, addr.sa_family, bAsync);
if (sockfd < 0) {
WarnL << "创建套接字失败:" << host;
return -1;
}
if (::connect(sockfd, &addr, sizeof(struct sockaddr)) == 0) {
//同步连接成功
return sockfd;
}
if (bAsync && get_uv_error(true) == UV_EAGAIN) {
//异步连接成功
return sockfd;
}
WarnL << "连接主机失败:" << host << " " << port << " " << get_uv_errmsg(true);
close(sockfd);
return -1;
}
Currently, the SockUtil::connect function creates and binds the socket, then immediately connects to the remote destination IP and port. However, sometimes these two things need to be done separately. For example, during GB28181 playback negotiation, when the subordinate receives the invite request from the superior and returns a 200ok SDP message with the streaming source IP and port, some superiors have added security boundary devices that require verification of the subordinate's source IP and port before allowing the subordinate to connect and send stream data. In this case, three steps are required: 1. Create and bind the local port; 2. Send the streaming IP and port to the superior and wait for the superior's ack; 3. Connect to the remote destination IP and port. The specific implementation code is:
TRANS_BY_GITHUB_AI_ASSISTANT