diff options
| -rw-r--r-- | include/sibs/DirectConnection.hpp | 3 | ||||
| -rw-r--r-- | src/BootstrapConnection.cpp | 2 | ||||
| -rw-r--r-- | src/DirectConnection.cpp | 32 |
3 files changed, 35 insertions, 2 deletions
diff --git a/include/sibs/DirectConnection.hpp b/include/sibs/DirectConnection.hpp index 0ac5724..650486e 100644 --- a/include/sibs/DirectConnection.hpp +++ b/include/sibs/DirectConnection.hpp @@ -51,6 +51,9 @@ namespace sibs void connectServer(const Ipv4 &address, PubSubConnectCallback connectCallbackFunc, PubSubReceiveDataCallback receiveDataCallbackFunc); // Throws ConnectionException on error void connect(const Ipv4 &address, PubSubConnectCallback connectCallbackFunc, PubSubReceiveDataCallback receiveDataCallbackFunc); + + void connect(const Ipv4 &address, const std::shared_ptr<DirectConnectionPeer> serverPeer, PubSubConnectCallback connectCallbackFunc, PubSubReceiveDataCallback receiveDataCallbackFunc); + void send(const std::shared_ptr<DirectConnectionPeer> peer, std::shared_ptr<std::vector<u8>> data, PubSubSendDataCallback sendDataCallbackFunc = nullptr); protected: int createSocket(const Ipv4 &addressToBind, bool rendezvous, bool reuseAddr, bool bind = true); diff --git a/src/BootstrapConnection.cpp b/src/BootstrapConnection.cpp index 9a19943..924fcff 100644 --- a/src/BootstrapConnection.cpp +++ b/src/BootstrapConnection.cpp @@ -61,7 +61,7 @@ namespace sibs newPeerAddress.address.sin_addr.s_addr = ipv4Address; newPeerAddress.address.sin_port = port; memset(newPeerAddress.address.sin_zero, 0, sizeof(newPeerAddress.address.sin_zero)); - connections.connect(newPeerAddress, [this, pubsubKey](std::shared_ptr<DirectConnectionPeer> newPeer, PubSubResult result, const std::string &resultStr) + connections.connect(newPeerAddress, peer, [this, pubsubKey](std::shared_ptr<DirectConnectionPeer> newPeer, PubSubResult result, const std::string &resultStr) { if(result == PubSubResult::OK) { diff --git a/src/DirectConnection.cpp b/src/DirectConnection.cpp index 36d2d16..c19fa63 100644 --- a/src/DirectConnection.cpp +++ b/src/DirectConnection.cpp @@ -75,13 +75,43 @@ namespace sibs void DirectConnections::connectServer(const Ipv4 &address, PubSubConnectCallback connectCallbackFunc, PubSubReceiveDataCallback receiveDataCallbackFunc) { - connect(address, false, true, connectCallbackFunc, receiveDataCallbackFunc, false); + connect(address, true, true, connectCallbackFunc, receiveDataCallbackFunc, true); } void DirectConnections::connect(const Ipv4 &address, PubSubConnectCallback connectCallbackFunc, PubSubReceiveDataCallback receiveDataCallbackFunc) { connect(address, true, true, connectCallbackFunc, receiveDataCallbackFunc, true); } + + void DirectConnections::connect(const Ipv4 &address, std::shared_ptr<DirectConnectionPeer> serverPeer, PubSubConnectCallback connectCallbackFunc, PubSubReceiveDataCallback receiveDataCallbackFunc) + { + std::thread([this, address, connectCallbackFunc, receiveDataCallbackFunc, serverPeer]() + { + if(UDT::bind(serverPeer->socket, (sockaddr*)&address.address, sizeof(address.address)) == UDT::ERROR) + { + std::string errMsg = "UDT: Failed to bind, error: "; + errMsg += UDT::getlasterror_desc(); + throw SocketCreateException(errMsg); + } + + Log::debug("DirectConnections: Connecting to peer (ip: %s, port: %d, rendezvous: yes)", address.getAddress().c_str(), address.getPort()); + if(UDT::connect(serverPeer->socket, (sockaddr*)&address.address, sizeof(address.address)) == UDT::ERROR) + { + if(connectCallbackFunc) + connectCallbackFunc(serverPeer, PubSubResult::ERROR, UDT::getlasterror_desc()); + return; + } + + serverPeer->address = address; + serverPeer->receiveDataCallbackFunc = receiveDataCallbackFunc; + peersMutex.lock(); + peers[serverPeer->socket] = serverPeer; + peersMutex.unlock(); + + if(connectCallbackFunc) + connectCallbackFunc(serverPeer, PubSubResult::OK, ""); + }).detach(); + } void DirectConnections::connect(const Ipv4 &address, bool rendezvous, bool reuseAddr, PubSubConnectCallback connectCallbackFunc, PubSubReceiveDataCallback receiveDataCallbackFunc, bool bind) { |