mirrors/opendht
1
0 Fork 0
mirror of https://github.com/savoirfairelinux/opendht.git synced 2026-11-08 04:21:51 +00:00
Code Issues Packages Projects Releases Wiki Activity

wip

Browse Source
This commit is contained in:
Adrien Beraud authored and Adrien Béraud committed 2017-10-17 19:00:53 +02:00
1 parent 2acdc1f1aa
commit 419fc24f1f
4 files changed
+65 -33
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

include/opendht/network_engine.h
+4 -3
View File
@@ -479,10 +479,10 @@ private:
};
int send(msgpack::sbuffer& msg, const Sp<Node>& node);
int send(msgpack::sbuffer& msg, const Node& node, const Sp<TcpSocket>& sock = {});
int sendUDP(msgpack::sbuffer& msg, const SockAddr& addr);
int send(const Blob& msg, int flags, const Sp<Node>& node, UdpSocket::OnSent&& cb);
int send(const Blob& msg, int flags, const Node& node, UdpSocket::OnSent&& cb);
void startTcp(const Sp<TcpSocket>& sock, const Sp<Node>& node = {});
@@ -519,7 +519,8 @@ private:
TransId tid,
uint16_t code,
const std::string& message,
bool include_id=false);
bool include_id=false,
const Sp<TcpSocket>& sock= {});
void deserializeNodes(ParsedMessage& msg, const SockAddr& from);
include/opendht/node.h
+7 -2
View File
@@ -53,7 +53,7 @@ struct Node {
Sp<TcpSocket> sock;
InfoHash last_known_pk;
//std::list<crypto::EcSecretKey>
//std::list<crypto::EcSecretKey>
Node(const InfoHash& id, const SockAddr& addr, const Sp<TcpSocket>& s = {}, bool client=false);
Node(const InfoHash& id, const sockaddr* sa, socklen_t salen)
@@ -70,6 +70,7 @@ struct Node {
std::string getAddrStr() const {
return addr.toString();
}
bool isClient() const { return is_client; }
bool isIncoming() { return time > reply_time; }
@@ -77,8 +78,12 @@ struct Node {
const time_point& getReplyTime() const { return reply_time; }
void setTime(const time_point& t) { time = t; }
SockAddr getAddress() const {
return canStream() ? sock->getPeerAddr() : addr;
}
void startTCP() {
}
/**
src/dht.cpp
+1 -1
View File
@@ -2237,7 +2237,7 @@ Dht::onListen(Sp<Node> node, const InfoHash& hash, const Blob& token, size_t soc
net::DhtProtocolException::LISTEN_NO_INFOHASH
};
}
if (not tokenMatch(token, node->getAddr())) {
if (not tokenMatch(token, node->getAddress())) {
DHT_LOG.w(hash, node->id, "[node %s] incorrect token %s for 'listen'", node->toString().c_str(), hash.toString().c_str());
throw net::DhtProtocolException {net::DhtProtocolException::UNAUTHORIZED, net::DhtProtocolException::LISTEN_WRONG_TOKEN};
}
src/network_engine.cpp
+53 -27
View File
@@ -165,6 +165,20 @@ NetworkEngine::requestStep(Sp<Request> sreq)
req.on_expired(req, false);
}
if (node.sock and node.sock->canWrite()) {
auto data = (uint8_t*)malloc(req.msg.size());
memcpy(data, req.msg.data(), req.msg.size());
req.last_try = now;
++req.attempt_count;
node.sock->write(data, req.msg.size(), [&,sreq](int status){
if (status != 0) {
DHT_LOG.w(sreq->node->id, "write failed ! %d", status);
sreq->node->setExpired();
}
});
return;
}
// UDP
std::weak_ptr<Request> wreq = sreq;
auto err = send(req.msg, 0, node.getAddr(), [this,wreq](int status) {
if (auto req = wreq.lock()) {
@@ -291,8 +305,10 @@ NetworkEngine::startTcp(const Sp<TcpSocket>& sock, const Sp<Node>& node)
msgpack::object_handle result;
// Message pack data loop
while(rx_data->unpacker.next(result)) {
//std::unique_ptr<ParsedMessage> msg {new ParsedMessage};
std::unique_ptr<ParsedMessage> msg {new ParsedMessage};
SockAddr from = sock->getPeerAddr();
if (from.isMappedIPv4())
from = from.getMappedIPv4();
Sp<Node> node_found;
try {
const auto& o = result.get();
@@ -302,8 +318,8 @@ NetworkEngine::startTcp(const Sp<TcpSocket>& sock, const Sp<Node>& node)
//msg->msgpack_unpack(oh.get());
node_found = processEncrypted(oh.get(), from);
} else {
//msg->msgpack_unpack(o);
node_found = process(o, from);
msg->msgpack_unpack(o);
node_found = process(std::move(msg), from, sock);
}
} catch (const std::exception& e) {
DHT_LOG.w("Can't process message: %s", e.what());
@@ -452,10 +468,10 @@ NetworkEngine::process(const msgpack::object& o, const SockAddr& from)
}
Sp<Node>
NetworkEngine::process(std::unique_ptr<ParsedMessage>&& msg, const SockAddr& from, const Sp<TcpSocket>& sock)
NetworkEngine::process(std::unique_ptr<ParsedMessage>&& msg, const SockAddr& from, const Sp<TcpSocket>& tcpsock)
{
const auto& now = scheduler.time();
auto node = cache.getNode(msg->id, from, now, sock, true, msg->is_client);
auto node = cache.getNode(msg->id, from, now, tcpsock, true, msg->is_client);
if (msg->type == MessageType::Error or msg->type == MessageType::Reply) {
auto rsocket = node->getSocket(msg->tid.toInt());
@@ -476,7 +492,7 @@ NetworkEngine::process(std::unique_ptr<ParsedMessage>&& msg, const SockAddr& fro
return;
}
}
node->update(from, sock);
node->update(from, tcpsock);
node->received(now, req);
@@ -582,7 +598,8 @@ NetworkEngine::process(std::unique_ptr<ParsedMessage>&& msg, const SockAddr& fro
} catch (const std::overflow_error& e) {
DHT_LOG.e("Can't send value: buffer not large enough !");
} catch (DhtProtocolException& e) {
sendError(node, msg->tid, e.getCode(), e.getMsg().c_str(), true);
DHT_LOG.w("Error during request handling, sending error ! %p", tcpsock.get());
sendError(node, msg->tid, e.getCode(), e.getMsg().c_str(), true, tcpsock);
}
}
@@ -617,17 +634,18 @@ NetworkEngine::sendUDP(msgpack::sbuffer& msg, const SockAddr& addr)
}
int
NetworkEngine::send(msgpack::sbuffer& msg, const Sp<Node>& node)
NetworkEngine::send(msgpack::sbuffer& msg, const Node& node, const Sp<TcpSocket>& tcpsock)
{
DHT_LOG.d(node.id, "[node %s] send %lu", node.getId().to_c_str(), msg.size());
uint8_t* data;
size_t size;
if (node->id == InfoHash{}) {
if (node.id == InfoHash{}) {
size = msg.size();
data = (uint8_t*)msg.release();
} else {
Blob encrypted_out;
{
Blob encrypted = id_key.encrypt((const uint8_t *)msg.data(), msg.size(), node->id);
Blob encrypted = id_key.encrypt((const uint8_t *)msg.data(), msg.size(), node.id);
msgpack::sbuffer buffer;
msgpack::packer<msgpack::sbuffer> pk(&buffer);
pk.pack_array(2);
@@ -635,7 +653,7 @@ NetworkEngine::send(msgpack::sbuffer& msg, const Sp<Node>& node)
pk.pack_bin(encrypted.size());
pk.pack_bin_body((const char *)encrypted.data(), encrypted.size());
encrypted_out = node->id.encrypt((const uint8_t *)buffer.data(), buffer.size());
encrypted_out = node.id.encrypt((const uint8_t *)buffer.data(), buffer.size());
}
msgpack::sbuffer buffer;
msgpack::packer<msgpack::sbuffer> pk(&buffer);
@@ -644,26 +662,31 @@ NetworkEngine::send(msgpack::sbuffer& msg, const Sp<Node>& node)
size = buffer.size();
data = (uint8_t*)buffer.release();
}
if (node->canStream()) {
return node->sock->write(data, size);
if (tcpsock) {
return tcpsock->write(data, size);
} else if (node.canStream()) {
DHT_LOG.d(node.id, "[node %s] send (stream) %lu", node.getId().to_c_str(), msg.size());
return node.sock->write(data, size);
} else {
return sock->send(data, size, node->addr);
DHT_LOG.d(node.id, "[node %s] send (datagram) %lu", node.getId().to_c_str(), msg.size());
return sock->send(data, size, node.addr);
}
}
int
NetworkEngine::send(const Blob& msg, int /*flags*/, const Sp<Node>& node, UdpSocket::OnSent&& cb)
NetworkEngine::send(const Blob& msg, int /*flags*/, const Node& node, UdpSocket::OnSent&& cb)
{
DHT_LOG.d(node.id, "[node %s] send %lu", node.getId().to_c_str(), msg.size());
uint8_t* data;
size_t size;
if (node->id == InfoHash{}) {
if (node.id == InfoHash{}) {
size = msg.size();
data = (uint8_t*)malloc(size);//(uint8_t*)msg.release();
std::memcpy(data, msg.data(), size);
} else {
Blob encrypted_out;
{
Blob encrypted = id_key.encrypt((const uint8_t *)msg.data(), msg.size(), node->id);
Blob encrypted = id_key.encrypt((const uint8_t *)msg.data(), msg.size(), node.id);
msgpack::sbuffer buffer;
msgpack::packer<msgpack::sbuffer> pk(&buffer);
pk.pack_array(2);
@@ -671,7 +694,7 @@ NetworkEngine::send(const Blob& msg, int /*flags*/, const Sp<Node>& node, UdpSoc
pk.pack_bin(encrypted.size());
pk.pack_bin_body((const char *)encrypted.data(), encrypted.size());
encrypted_out = node->id.encrypt((const uint8_t *)buffer.data(), buffer.size());
encrypted_out = node.id.encrypt((const uint8_t *)buffer.data(), buffer.size());
}
msgpack::sbuffer buffer;
msgpack::packer<msgpack::sbuffer> pk(&buffer);
@@ -680,10 +703,12 @@ NetworkEngine::send(const Blob& msg, int /*flags*/, const Sp<Node>& node, UdpSoc
size = buffer.size();
data = (uint8_t*)buffer.release();
}
if (node->canStream()) {
return node->sock->write(data, size);
if (node.canStream()) {
DHT_LOG.d(node.id, "[node %s] send (stream) %lu", node.getId().to_c_str(), msg.size());
return node.sock->write(data, size);
} else {
return sock->send(data, size, node->addr, std::move(cb));
DHT_LOG.d(node.id, "[node %s] send (datagram) %lu", node.getId().to_c_str(), msg.size());
return sock->send(data, size, node.addr, std::move(cb));
}
}
@@ -743,7 +768,7 @@ NetworkEngine::sendPong(const Sp<Node>& node, TransId tid) {
pk.pack(std::string("n")); pk.pack(network);
}
send(buffer, node);
send(buffer, *node);
}
Sp<Request>
@@ -1003,7 +1028,7 @@ NetworkEngine::sendNodesValues(const Sp<Node>& node, TransId tid, const Blob& no
}
// send response
send(buffer, node);
send(buffer, *node);
// send parts
if (not svals.empty())
@@ -1154,7 +1179,7 @@ NetworkEngine::sendListenConfirmation(const Sp<Node>& node, TransId tid) {
pk.pack(std::string("n")); pk.pack(network);
}
send(buffer, node);
send(buffer, *node);
}
Sp<Request>
@@ -1287,7 +1312,7 @@ NetworkEngine::sendValueAnnounced(const Sp<Node>& node, TransId tid, Value::Id v
pk.pack(std::string("n")); pk.pack(network);
}
send(buffer, node);
send(buffer, *node);
}
void
@@ -1295,7 +1320,8 @@ NetworkEngine::sendError(const Sp<Node>& node,
TransId tid,
uint16_t code,
const std::string& message,
bool include_id)
bool include_id,
const Sp<TcpSocket>& tcpsock)
{
msgpack::sbuffer buffer;
msgpack::packer<msgpack::sbuffer> pk(&buffer);
@@ -1318,7 +1344,7 @@ NetworkEngine::sendError(const Sp<Node>& node,
pk.pack(std::string("n")); pk.pack(network);
}
send(buffer, node);
send(buffer, *node, tcpsock);
}
void