import lbms.plugins.mldht.kad.DHT.DHTtype; //导入依赖的package包/类
/**
 * Packs a BucketEntry into the provided buffer.
 *
 * @param e Entry to insert
 * @param buffer must be at least 26bytes per Entry
 * @param off Offset to use
 * @throws IllegalArgumentException if buffer is too small
 */
public static void PackBucketEntry (KBucketEntry e, byte[] buffer, int off, DHTtype type) {
	// first check size
	if (off + type.NODES_ENTRY_LENGTH > buffer.length) {
		throw new IllegalArgumentException("Not enough room in buffer");
	}
	ByteBuffer bb = ByteBuffer.wrap(buffer, off, type.NODES_ENTRY_LENGTH);

	InetSocketAddress addr = e.getAddress();
	if(type == DHTtype.IPV6_DHT && addr.getAddress() instanceof Inet4Address)
		throw new IllegalArgumentException("Attempting to serialize an IPv4 bucket entry into nodes6 buffer");
	// copy ID, IP address and port into the buffer
	bb.put(e.getID().getHash());
	bb.put(addr.getAddress().getAddress());
	//bt::WriteUint32(ptr,20,addr.ipAddress().IPv4Addr());
	bb.putShort((short) addr.getPort());
}
 

import lbms.plugins.mldht.kad.DHT.DHTtype; //导入依赖的package包/类
/**
 * Unpacks a Entry from a byte array
 *
 * @param buffer byte array with serialized entry
 * @param off the Offset to use
 * @return deserialized Entry
 * @throws IllegalArgumentException if buffer is to small
 */
public static KBucketEntry UnpackBucketEntry (byte[] buffer, int off, DHTtype type) {
	if (off + type.NODES_ENTRY_LENGTH > buffer.length) {
		throw new IllegalArgumentException("Not enough room in buffer");
	}
	ByteBuffer bb = ByteBuffer.wrap(buffer, off, type.NODES_ENTRY_LENGTH);

	byte[] key = new byte[20];
	bb.get(key);

	byte[] inetaddr = new byte[type.NODES_ENTRY_LENGTH - 20 - 2];
	bb.get(inetaddr);

	InetSocketAddress addr = null;
	//UnknownHostException shouldn't occur since IP is provided
	try {

		addr = new InetSocketAddress(InetAddress.getByAddress(inetaddr), Short.toUnsignedInt(bb.getShort()));
	} catch (UnknownHostException e) {
		e.printStackTrace();
		throw new RuntimeException(e);
	}

	return new KBucketEntry(addr, new Key(key), 0);
}
 

import lbms.plugins.mldht.kad.DHT.DHTtype; //导入依赖的package包/类
public NodeList asNodeList() {
	return new NodeList() {
		
		@Override
		public int packedSize() {
			return entries.size() * owner.getType().NODES_ENTRY_LENGTH;
		}
		
		@Override
		public Stream<KBucketEntry> entries() {
			return entries.stream();
		}

		@Override
		public AddressType type() {
			return owner.getType() == DHTtype.IPV4_DHT ? AddressType.V4 : AddressType.V6;
		}
	};
}
 

import lbms.plugins.mldht.kad.DHT.DHTtype; //导入依赖的package包/类
StorageItem roundTripPut(StorageItem in) throws IOException, MessageException {
	// create message from item
	PutRequest req = new PutRequest();
	req.populateFromStorage(in);
	req.setID(Key.createRandomKey());
	req.setMTID(new byte[4]);
	req.setToken(new byte[4]);

	// encode
	ByteBuffer encoded = ByteBuffer.allocate(1500);
	req.encode(encoded);
	
	// decode
	Map<String, Object> decoded = new BDecoder().decode(encoded.duplicate());
	MessageDecoder decoder = new MessageDecoder(null, DHTtype.IPV4_DHT);
	decoder.toDecode(encoded, decoded);
	PutRequest roundTripped = (PutRequest) decoder.parseMessage();
	
	// re-create item from round-tripped message
	return new StorageItem(roundTripped);
}
 

import lbms.plugins.mldht.kad.DHT.DHTtype; //导入依赖的package包/类
@Inject
public MldhtService(IRuntimeLifecycleBinder lifecycleBinder, Config config, DHTConfig dhtConfig) {
    this.dht = new DHT(dhtConfig.shouldUseIPv6()? DHTtype.IPV6_DHT : DHTtype.IPV4_DHT);
    this.config = toMldhtConfig(dhtConfig);
    this.localAddress = config.getAcceptorAddress();
    this.useRouterBootstrap = dhtConfig.shouldUseRouterBootstrap();
    this.publicBootstrapNodes = dhtConfig.getPublicBootstrapNodes();
    this.bootstrapNodes = dhtConfig.getBootstrapNodes();

    lifecycleBinder.onStartup(LifecycleBinding.bind(this::start).description("Initialize DHT facilities").async().build());
    lifecycleBinder.onShutdown("Shutdown DHT facilities", this::shutdown);
}
 

import lbms.plugins.mldht.kad.DHT.DHTtype; //导入依赖的package包/类
public InetAddress getInetAddress() {
	try
	{
		if (item.length == DHTtype.IPV4_DHT.ADDRESS_ENTRY_LENGTH)
			return InetAddress.getByAddress(Arrays.copyOf(item, 4));
		if (item.length == DHTtype.IPV6_DHT.ADDRESS_ENTRY_LENGTH)
			return InetAddress.getByAddress(Arrays.copyOf(item, 16));
	} catch (UnknownHostException e)
	{
		// should not happen
		e.printStackTrace();
	}
	
	return null;
}
 

import lbms.plugins.mldht.kad.DHT.DHTtype; //导入依赖的package包/类
public Class<? extends InetAddress> getAddressType() {
	if(item.length == DHTtype.IPV4_DHT.ADDRESS_ENTRY_LENGTH)
		return DHTtype.IPV4_DHT.PREFERRED_ADDRESS_TYPE;
	if(item.length == DHTtype.IPV6_DHT.ADDRESS_ENTRY_LENGTH)
		return DHTtype.IPV6_DHT.PREFERRED_ADDRESS_TYPE;
	return null;
}
 

import lbms.plugins.mldht.kad.DHT.DHTtype; //导入依赖的package包/类
public int getPort() {
	if (item.length == DHTtype.IPV4_DHT.ADDRESS_ENTRY_LENGTH)
		return (item[4] & 0xFF) << 8 | (item[5] & 0xFF);
	if (item.length == DHTtype.IPV6_DHT.ADDRESS_ENTRY_LENGTH)
		return (item[16] & 0xFF) << 8 | (item[17] & 0xFF);
	return 0;
}
 

import lbms.plugins.mldht.kad.DHT.DHTtype; //导入依赖的package包/类
@Override
void update () {
	// go over the todo list and send find node calls
	// until we have nothing left

		while (canDoRequest()) {
			synchronized (todo) {

			KBucketEntry e = todo.stream().findAny().orElse(null);
			if(e == null)
				break;
			
			if (visited.contains(e.getAddress()))
				continue;
			
			// send a findNode to the node
			FindNodeRequest fnr;

			fnr = new FindNodeRequest(Key.createRandomKey());
			fnr.setWant4(rpc.getDHT().getType() == DHTtype.IPV4_DHT);
			fnr.setWant6(rpc.getDHT().getType() == DHTtype.IPV6_DHT);
			fnr.setDestination(e.getAddress());
			rpcCall(fnr,e.getID(), c -> {
				todo.remove(e);
				visited.add(e.getAddress());
			});
		}
	}
}
 

import lbms.plugins.mldht.kad.DHT.DHTtype; //导入依赖的package包/类
@Override
void update () {
	for(;;) {
		// while(!todo.isEmpty() && canDoRequest() && !isClosestSetStable() && !nextTodoUseless())
		RequestPermit p = checkFreeSlot();
		if(p == RequestPermit.NONE_ALLOWED)
			return;
		
		KBucketEntry e = todo.next().orElse(null);
		
		if(e == null)
			return;
		
		if(!new RequestCandidateEvaluator(this, closest, todo, e, inFlight).goodForRequest(p))
			return;
			
		// send a findNode to the node
		FindNodeRequest fnr = new FindNodeRequest(targetKey);
		fnr.setWant4(rpc.getDHT().getType() == DHTtype.IPV4_DHT || rpc.getDHT().getSiblings().stream().anyMatch(sib -> sib.isRunning() && sib.getType() == DHTtype.IPV4_DHT && sib.getNode().getNumEntriesInRoutingTable() < DHTConstants.BOOTSTRAP_IF_LESS_THAN_X_PEERS));
		fnr.setWant6(rpc.getDHT().getType() == DHTtype.IPV6_DHT || rpc.getDHT().getSiblings().stream().anyMatch(sib -> sib.isRunning() && sib.getType() == DHTtype.IPV6_DHT && sib.getNode().getNumEntriesInRoutingTable() < DHTConstants.BOOTSTRAP_IF_LESS_THAN_X_PEERS));
		fnr.setDestination(e.getAddress());
		
		if(!rpcCall(fnr,e.getID(), (call) -> {
			long rtt = e.getRTT();
			rtt = rtt + rtt / 2; // *1.5 since this is the average and not the 90th percentile like the timeout filter
			if(rtt < DHTConstants.RPC_CALL_TIMEOUT_MAX && rtt < rpc.getTimeoutFilter().getStallTimeout())
				call.setExpectedRTT(rtt); // only set a node-specific timeout if it's better than what the server would apply anyway
			call.builtFromEntry(e);
			todo.addCall(call, e);
			
			if(DHT.isLogLevelEnabled(LogLevel.Verbose)) {
				List<InetSocketAddress> sources = todo.getSources(e).stream().map(KBucketEntry::getAddress).collect(Collectors.toList());
				DHT.log("Task "+getTaskID()+" sending call to "+ e + " sources:" + sources, LogLevel.Verbose);
			}
		})) {
			break;
		}
			
	}
}
 

import lbms.plugins.mldht.kad.DHT.DHTtype; //导入依赖的package包/类
@Override
void callFinished (RPCCall c, MessageBase rsp) {

	// check the response and see if it is a good one
	if (rsp.getMethod() != Method.FIND_NODE || rsp.getType() != Type.RSP_MSG)
		return;
	
	KBucketEntry match = todo.acceptResponse(c);
	
	if(match == null)
		return;

	FindNodeResponse fnr = (FindNodeResponse) rsp;
	
	closest.insert(match);
	
	
	for (DHTtype type : DHTtype.values())
	{
		NodeList nodes = fnr.getNodes(type);
		if (nodes == null)
			continue;
		if (type == rpc.getDHT().getType()) {
			Set<KBucketEntry> entries = nodes.entries().filter(e -> !AddressUtils.isBogon(e.getAddress()) && !node.isLocalId(e.getID())).collect(Collectors.toSet());
			todo.addCandidates(match, entries);
		} else {
			rpc.getDHT().getSiblings().stream().filter(sib -> sib.getType() == type).forEach(sib -> {
				nodes.entries().forEach(e -> {
					sib.addDHTNode(e.getAddress().getAddress().getHostAddress(), e.getAddress().getPort());
				});
			});
		}
	}

}
 

import lbms.plugins.mldht.kad.DHT.DHTtype; //导入依赖的package包/类
@Override
void update() {
	for(;;) {
		RequestPermit p = checkFreeSlot();
		if(p == RequestPermit.NONE_ALLOWED)
			return;
		
		KBucketEntry e = todo.next().orElse(null);
		if(e == null)
			return;
		
		if(!new RequestCandidateEvaluator(this, closest, todo, e, inFlight).goodForRequest(p))
			return;
			
		GetRequest r = new GetRequest(targetKey);
		
		r.setWant4(node.getDHT().getType() == DHTtype.IPV4_DHT);
		r.setWant6(node.getDHT().getType() == DHTtype.IPV6_DHT);
		r.setDestination(e.getAddress());
		if(expectedSequence != -1)
			r.setSeq(expectedSequence);
		
		if(!rpcCall(r, e.getID(), c -> {
			c.builtFromEntry(e);
			todo.addCall(c, e);
			int rtt = e.getRTT() * 2;
			if(rtt < DHTConstants.RPC_CALL_TIMEOUT_MAX && rtt < rpc.getTimeoutFilter().getStallTimeout())
				c.setExpectedRTT(rtt);
		})) {
			break;
		}
			
		
		
	}
}
 

import lbms.plugins.mldht.kad.DHT.DHTtype; //导入依赖的package包/类
public NodeList getNodes(DHTtype type)
{
	if(type == DHTtype.IPV4_DHT)
		return nodes;
	if(type == DHTtype.IPV6_DHT)
		return nodes6;
	return null;
}
 

import lbms.plugins.mldht.kad.DHT.DHTtype; //导入依赖的package包/类
@Override
public String toString() {
	return super.toString() +
		(nodes != null ? "contains: "+ (nodes.packedSize()/DHTtype.IPV4_DHT.NODES_ENTRY_LENGTH) + " nodes " : "") +
		(nodes6 != null ? "contains: "+ (nodes6.packedSize()/DHTtype.IPV6_DHT.NODES_ENTRY_LENGTH) + " nodes6 " : "") +
		(token != null ? "token "+token.length+" | " : "");
}
 

import lbms.plugins.mldht.kad.DHT.DHTtype; //导入依赖的package包/类
private Optional<NodeList> extractNodes(Map<String, Object> args, String key, DHTtype nodesType) throws MessageException {
	byte[] raw = typedGet(args, key, byte[].class).orElse(null);
	if(raw == null)
		return Optional.empty();
	if(raw.length % nodesType.NODES_ENTRY_LENGTH != 0)
		throw new MessageException("expected "+key+" length to be a multiple of "+nodesType.NODES_ENTRY_LENGTH+", received "+raw.length, ErrorCode.ProtocolError);
	return Optional.of(NodeList.fromBuffer(ByteBuffer.wrap(raw), nodesType == DHTtype.IPV4_DHT ? AddressType.V4 : AddressType.V6));
}
 

import lbms.plugins.mldht.kad.DHT.DHTtype; //导入依赖的package包/类
@Override
public String toString() {
	return super.toString() +
		(nodes != null ? (nodes.packedSize()/DHTtype.IPV4_DHT.NODES_ENTRY_LENGTH)+" nodes | " : "") +
		(nodes6 != null ? (nodes6.packedSize()/DHTtype.IPV6_DHT.NODES_ENTRY_LENGTH)+" nodes6 | " : "") +
		(items != null ? (items.size())+" values | " : "") +
		(scrapePeers != null ? "peer bloom filter | " : "") +
		(scrapeSeeds != null ? "seed bloom filter | " :  "" );
}
 

import lbms.plugins.mldht.kad.DHT.DHTtype; //导入依赖的package包/类
static DHT buildDHT(DHT.DHTtype type) {
	DHT dht = new DHT(type);
	dht.config = new DHTConfiguration() {
		
		@Override
		public boolean noRouterBootstrap() {
			// TODO Auto-generated method stub
			return false;
		}
		
		@Override
		public boolean isPersistingID() {
			// TODO Auto-generated method stub
			return false;
		}
		
		@Override
		public Path getStoragePath() {
			// TODO Auto-generated method stub
			return Paths.get(".", "does", "not", "exist");
		}
		
		@Override
		public int getListeningPort() {
			// TODO Auto-generated method stub
			return 0;
		}
		
		@Override
		public boolean allowMultiHoming() {
			// TODO Auto-generated method stub
			return false;
		}
	};
	dht.populate();
	
	return dht;
	
}
 

import lbms.plugins.mldht.kad.DHT.DHTtype; //导入依赖的package包/类
public PeerAddressDBItem(byte[] data, boolean isSeed) {
	super(data);
	if(data.length != DHTtype.IPV4_DHT.ADDRESS_ENTRY_LENGTH && data.length != DHTtype.IPV6_DHT.ADDRESS_ENTRY_LENGTH)
		throw new IllegalArgumentException("byte array length does not match ipv4 or ipv6 raw InetAddress+Port length");
	seed = isSeed;
}
 

import lbms.plugins.mldht.kad.DHT.DHTtype; //导入依赖的package包/类
@Override
void update () {
	// check if the cache has any closer nodes after the initial query
	if(useCache) {
		Collection<KBucketEntry> cacheResults = cache.get(targetKey, requestConcurrency());
		todo.addCandidates(null, cacheResults);
	}

	for(;;) {
		synchronized (this) {
			RequestPermit p = checkFreeSlot();
			
			if(p == RequestPermit.NONE_ALLOWED)
				break;
			
			KBucketEntry e = todo.next2(kbe -> {
				RequestCandidateEvaluator eval = new RequestCandidateEvaluator(this, closest, todo, kbe, inFlight);
				return eval.goodForRequest(p);
			}).orElse(null);
			
			if(e == null)
				break;
			
			GetPeersRequest gpr = new GetPeersRequest(targetKey);
			// we only request cross-seeding on find-node
			gpr.setWant4(rpc.getDHT().getType() == DHTtype.IPV4_DHT);
			gpr.setWant6(rpc.getDHT().getType() == DHTtype.IPV6_DHT);
			gpr.setDestination(e.getAddress());
			gpr.setScrape(scrapeHandler != null);
			gpr.setNoSeeds(noSeeds);
			
			if(!rpcCall(gpr, e.getID(), call -> {
				if(useCache)
					call.addListener(cache.getRPCListener());
				call.builtFromEntry(e);

				long rtt = e.getRTT();
				long defaultTimeout = rpc.getTimeoutFilter().getStallTimeout();
				
				if(rtt < DHTConstants.RPC_CALL_TIMEOUT_MAX) {
					// the measured RTT is a mean and not the 90th percentile unlike the RPCServer's timeout filter
					// -> add some safety margin to account for variance
					rtt = (long) (rtt * (rtt < defaultTimeout ? 2 : 1.5));
					
					call.setExpectedRTT(min(rtt, DHTConstants.RPC_CALL_TIMEOUT_MAX));
				}
				
				if(DHT.isLogLevelEnabled(LogLevel.Verbose)) {
					List<InetSocketAddress> sources = todo.getSources(e).stream().map(KBucketEntry::getAddress).collect(Collectors.toList());
					DHT.log("Task "+getTaskID()+" sending call to "+ e + " sources:" + sources, LogLevel.Verbose);
				}
					
				
				todo.addCall(call, e);
			})) {
				break;
			}
		}
	}

}
 

import lbms.plugins.mldht.kad.DHT.DHTtype; //导入依赖的package包/类
public MessageDecoder(Function<byte[], Optional<Method>> transactionIdMapper, DHTtype type) {
	this.transactionIdMapper = transactionIdMapper;
	this.type = type;
}