LBRYFoundation/lbry-globe-java
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Improve DHT

Browse source
This commit is contained in:
Ben van Hartingsveldt 2025-07-15 15:23:33 +02:00
parent 6fd1a3f426
commit 79fc5930ea
No known key found for this signature in database
GPG key ID: 261AA214130CE7AB
3 changed files with 39 additions and 36 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

20
src/main/java/com/lbry/globe/thread/DHTNodeFinderThread.java
View file

@ -38,7 +38,7 @@ public class DHTNodeFinderThread implements Runnable{
} }
this.startSender(); this.startSender();
this.startReceiver(); DHT.startReceiver();
} }
private void startSender(){ private void startSender(){
@ -120,22 +120,4 @@ public class DHTNodeFinderThread implements Runnable{
}).exceptionally((Throwable e) -> null); }).exceptionally((Throwable e) -> null);
} }
private void startReceiver(){
new Thread(() -> {
while(DHT.getSocket().isBound()) {
try {
UDP.Packet receiverPacket = UDP.receive(DHT.getSocket());
byte[] receivingBytes = receiverPacket.getData();
DHT.Message<?> message = DHT.Message.fromBencode(receivingBytes);
DHT.RPCID rpcid = new DHT.RPCID(message);
DHT.getFutureManager().finishFuture(rpcid,receiverPacket);
} catch (IOException e) {
e.printStackTrace();
}
}
},"DHT Receiver").start();
}
} }

4
src/main/java/com/lbry/globe/util/BencodeConverter.java
View file

@ -32,10 +32,10 @@ public class BencodeConverter{
} }
// Normal B-decoding // Normal B-decoding
return BencodeConverter.BENCODE.decode(bytes,Type.DICTIONARY); return BencodeConverter.walkAndConvertByteBufferToByteArrayOrString(BencodeConverter.BENCODE.decode(bytes,Type.DICTIONARY));
} }
public static <V> V walkAndConvertByteBufferToByteArrayOrString(Object value){ private static <V> V walkAndConvertByteBufferToByteArrayOrString(Object value){
if(value instanceof ByteBuffer){ if(value instanceof ByteBuffer){
ByteBuffer bb = (ByteBuffer) value; ByteBuffer bb = (ByteBuffer) value;
byte[] ba = bb.array(); byte[] ba = bb.array();

51
src/main/java/com/lbry/globe/util/DHT.java
View file

@ -38,30 +38,27 @@ public class DHT{
} }
public static CompletableFuture<UDP.Packet> ping(DatagramSocket socket,InetSocketAddress destination) throws IOException { public static CompletableFuture<UDP.Packet> ping(DatagramSocket socket,InetSocketAddress destination) throws IOException {
byte[] rpcID = new byte[20]; DHT.Message<String> pingMessage = new DHT.Message<>(DHT.Message.TYPE_REQUEST,DHT.RPCID.generate(),DHT.NODE_ID,"ping",Collections.singletonList(Collections.singletonMap("protocolVersion",1)));
new Random().nextBytes(rpcID);
DHT.Message<String> pingMessage = new DHT.Message<>(DHT.Message.TYPE_REQUEST,rpcID,DHT.NODE_ID,"ping",Collections.singletonList(Collections.singletonMap("protocolVersion",1)));
return DHT.sendWithFuture(socket,destination,pingMessage); return DHT.sendWithFuture(socket,destination,pingMessage);
} }
public static CompletableFuture<UDP.Packet> findNode(DatagramSocket socket,InetSocketAddress destination,byte[] key) throws IOException{ public static CompletableFuture<UDP.Packet> findNode(DatagramSocket socket,InetSocketAddress destination,byte[] key) throws IOException{
byte[] rpcID = new byte[20]; DHT.Message<String> findNodeMessage = new DHT.Message<>(DHT.Message.TYPE_REQUEST,DHT.RPCID.generate(),DHT.NODE_ID,"findNode",Arrays.asList(key,Collections.singletonMap("protocolVersion",1)));
new Random().nextBytes(rpcID);
DHT.Message<String> findNodeMessage = new DHT.Message<>(DHT.Message.TYPE_REQUEST,rpcID,DHT.NODE_ID,"findNode",Arrays.asList(key,Collections.singletonMap("protocolVersion",1)));
return DHT.sendWithFuture(socket,destination,findNodeMessage); return DHT.sendWithFuture(socket,destination,findNodeMessage);
} }
public static CompletableFuture<UDP.Packet> findValue(DatagramSocket socket,InetSocketAddress destination,byte[] key) throws IOException{ public static CompletableFuture<UDP.Packet> findValue(DatagramSocket socket,InetSocketAddress destination,byte[] key) throws IOException{
byte[] rpcID = new byte[20]; DHT.Message<String> findValueMessage = new DHT.Message<>(DHT.Message.TYPE_REQUEST,DHT.RPCID.generate(),DHT.NODE_ID,"findValue",Arrays.asList(key,Collections.singletonMap("protocolVersion",1)));
new Random().nextBytes(rpcID);
DHT.Message<String> findNodeMessage = new DHT.Message<>(DHT.Message.TYPE_REQUEST,rpcID,DHT.NODE_ID,"findValue",Arrays.asList(key,Collections.singletonMap("protocolVersion",1))); return DHT.sendWithFuture(socket,destination,findValueMessage);
}
return DHT.sendWithFuture(socket,destination,findNodeMessage); public static CompletableFuture<UDP.Packet> store(DatagramSocket socket,InetSocketAddress destination) throws IOException{
DHT.Message<String> storeMessage = new DHT.Message<>(DHT.Message.TYPE_REQUEST,DHT.RPCID.generate(),DHT.NODE_ID,"store");
return DHT.sendWithFuture(socket,destination,storeMessage);
} }
protected static CompletableFuture<UDP.Packet> sendWithFuture(DatagramSocket socket,InetSocketAddress destination, DHT.Message<?> message) throws IOException{ protected static CompletableFuture<UDP.Packet> sendWithFuture(DatagramSocket socket,InetSocketAddress destination, DHT.Message<?> message) throws IOException{
@ -144,12 +141,12 @@ public class DHT{
private DHT.Message<P> setFromBencode(byte[] data){ private DHT.Message<P> setFromBencode(byte[] data){
Map<String,?> dictionary = BencodeConverter.decode(data); Map<String,?> dictionary = BencodeConverter.decode(data);
this.type = ((Long) dictionary.get("0")).intValue(); this.type = ((Long) dictionary.get("0")).intValue();
this.rpcID = ((ByteBuffer) dictionary.get("1")).array(); this.rpcID = (byte[]) dictionary.get("1");
this.nodeID = ((ByteBuffer) dictionary.get("2")).array(); this.nodeID = (byte[]) dictionary.get("2");
this.payload = null; this.payload = null;
if(dictionary.containsKey("3")){ if(dictionary.containsKey("3")){
Object payload = dictionary.get("3"); Object payload = dictionary.get("3");
this.payload = BencodeConverter.walkAndConvertByteBufferToByteArrayOrString(payload); this.payload = (P) payload;
} }
this.arguments = null; this.arguments = null;
if(dictionary.containsKey("4")){ if(dictionary.containsKey("4")){
@ -175,6 +172,24 @@ public class DHT{
} }
public static void startReceiver(){
new Thread(() -> {
while(DHT.getSocket().isBound()) {
try {
UDP.Packet receiverPacket = UDP.receive(DHT.getSocket());
byte[] receivingBytes = receiverPacket.getData();
DHT.Message<?> message = DHT.Message.fromBencode(receivingBytes);
DHT.RPCID rpcid = new DHT.RPCID(message);
DHT.getFutureManager().finishFuture(rpcid,receiverPacket);
} catch (IOException e) {
e.printStackTrace();
}
}
},"DHT Receiver").start();
}
public static class RPCID{ public static class RPCID{
private final byte[] id; private final byte[] id;
@ -208,6 +223,12 @@ public class DHT{
'}'; '}';
} }
public static byte[] generate(){
byte[] rpcID = new byte[20];
new Random().nextBytes(rpcID);
return rpcID;
}
} }
} }