aboutsummaryrefslogtreecommitdiff
diff options
context:
space:
mode:
-rw-r--r--include/sibs/DirectConnection.hpp3
-rw-r--r--src/BootstrapConnection.cpp2
-rw-r--r--src/DirectConnection.cpp32
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)
{