# aardio 范例: DHT 客户端



```aardio
import win.ui;
import win.debounce;
import wsock.bt;
import thread.works;
/*DSG{{*/
var winform = win.form(text="DHT 客户端";right=828;bottom=574)
winform.add(
editDhtNodes={cls="edit";left=14;top=16;right=812;bottom=384;db=1;dl=1;dr=1;dt=1;edge=1;multiline=1;vscroll=1;z=1};
editMetadata={cls="edit";left=14;top=389;right=813;bottom=563;db=1;dl=1;dr=1;edge=1;multiline=1;vscroll=1;z=2};
splitter={cls="splitter";left=14;top=384;right=812;bottom=389;frame=1;horz=1;z=3}
)
/*}}*/

// DHT 路由表
var routingTable = wsock.bt.kBucket("/dht.dat");

// DHT 客户端
var dhtClient,err = wsock.bt.dhtClient("6881-7999",routingTable);
assert(dhtClient,err)

// 多线程下载元数据
var metadataWorker = thread.works( 8,
	function(peersInfo,winform) {
		
		import wsock.bt.metadataClient;
		import bencode; 
		
		var exClient = wsock.bt.metadataClient(peersInfo.infoHash,peersInfo.nodeId);
		for(i=1;#peersInfo){ 
			var metadata,err = exClient.getMetadata(peersInfo[i].ip,peersInfo[i].port);
				
			if(metadata){  
				
				var metaInfo = bencode.decode(metadata); 
				bencode.save("/"+string.hex(peersInfo.infoHash)+".torrent", {info=metaInfo})
				
				//可重写为保存到数据库
				..io.print(i,"获取 torrent 成功:", metaInfo.name);
				
				winform.onGetMetadataComplete(peersInfo.infoHash,true);
				return; // 成功后即返回，不再尝试其他 peer
			}
			else{
				..io.stderr.write(err,string.hex(peersInfo.infoHash),'\n')
			}	
		}
		
		winform.onGetMetadataComplete(peersInfo.infoHash,false);
	},/*waitTime*/,/*init*/,winform
);

/* 
info hash 队列。
多 DHT 客户端可修改为从数据库服务端存取。
*/
var infoHashQueue = {
	count = 0;
	hashes = {};
	getPeersAttempts = {};
	getPeersQueue = [];
	
	add = function(infoHash){ 
		if(owner.hashes[infoHash]){
			return;
		}
		
		owner.hashes[infoHash] = "lookup";
		owner.getPeersAttempts[infoHash] = 0;
		table.push(owner.getPeersQueue,infoHash);

		winform.editMetadata.print("发现 Magnet 链接:"
			, "magnet:?xt=urn:btih:" + string.hex(infoHash));
			
		owner.count = owner.count + 1;
		return true;
	};
	
	maxAttempts = 20;
	next = function(){  
		var q = owner.getPeersQueue;
		var infoHash = q[1];
		if(!infoHash) return;
			
		if(
			owner.getPeersAttempts[infoHash]>=owner.maxAttempts 
				&& (#q > 1) ){
		
			..table.shift(q);
			..table.push(q,infoHash)
				
			infoHash = q[1];
		}
		
		owner.hashes[infoHash] = "busy";
		owner.getPeersAttempts[infoHash] = owner.getPeersAttempts[infoHash] + 1;
		return infoHash,owner.getPeersAttempts[infoHash];
	};
	
	queriedNodes = ..table.cache();
	getClosestNodes = function(infoHash){
			var queriedNodes = owner.queriedNodes[infoHash]
			if(!queriedNodes){
				queriedNodes = {}
				owner.queriedNodes[infoHash] = queriedNodes;
			}
			
			//刷新最接近节点
			var closestNodes = routingTable.getClosestNodes(infoHash,50);
			
			var result = [];
			
			var limit = 0;
			
			//排除已查询节点
			for(i, node in closestNodes) { 
				if(!queriedNodes[node.nodeId]){
				queriedNodes[node.nodeId] = 1;
				table.push(result,node);
				
				limit++;
				
				if(limit>8) return result; //避免一次发送过多请求
			}
		}
		
		return result; 
	}; 	
	
	peersCache = {};
	delayedMetadataFetch = {};
	onPeersDiscovered = function(infoHash,peers){ 
		
		// 合并同一个 info hash 的 peers
		owner.peersCache[infoHash] = table.append(owner.peersCache[infoHash]||[],peers);
		
		// 延时下载
		if(!owner.delayedMetadataFetch[infoHash]){
			var this = owner;
			owner.delayedMetadataFetch[infoHash] = win.debounce( 
				function(){ 
					var peers = this.peersCache[infoHash];
					peers.nodeId = dhtClient.nodeId
					peers.infoHash = infoHash;
					
					// 发送给下载线程 
					metadataWorker.push( peers );
						
					winform.editMetadata.print("downloading/peers:"+#peers,string.hex(infoHash));
					this.peersCache[infoHash] = null;
				},10000
			); 
		}
		
		owner.delayedMetadataFetch[infoHash]();
		
		// 停止发送 get peers ，切换到下一个 infoHash
		owner.stopGetPeers(infoHash);
	};
	stopGetPeers = function(infoHash){
		var idx = table.removeByValue(owner.getPeersQueue,infoHash);
		
		owner.getPeersAttempts[infoHash] = 0;
		owner.hashes[infoHash] = "metadata";
	};
	complete = function(infoHash,success){
		owner.hashes[infoHash] = success ? "completed" : "failed";;
	};
}

winform.onGetMetadataComplete = function(infoHash,success){
	winform.editMetadata.print(success?"下载成功":"下载失败",string.hex(infoHash));
	infoHashQueue.complete(infoHash,success);
}

// 退出保存 DHT 路由表
winform.onDestroy = function(){  
	routingTable.save(); 
	metadataWorker.quit(5000);
}

dhtClient.onGetPeers = lambda (infoHash) infoHashQueue.add(infoHash); 
dhtClient.onAnnouncePeer = lambda (infoHash) infoHashQueue.add(infoHash); 
	
var onNodesOrPeersDiscovered = function(nodes,peers,ip,port,infoHash){
	if(peers){
		infoHashQueue.onPeersDiscovered(infoHash,peers);
	}
	elseif(nodes){ 
		for i,node in nodes{
			if routingTable.addNode(node) {
					winform.editDhtNodes.print("发现新节点:"
						, node.ip, node.port, "总节点数:", routingTable.getNodeCount());
			} 
		}  
	} 	
}

// 定时查找可以获取元数据的 peers
winform.setInterval( 
	function(){ 
		
		var infoHash,attempts  = infoHashQueue.next()
		if(!infoHash) return;
		
		io.print("get peers:",attempts,string.hex(infoHash))

		var closestNodes = infoHashQueue.getClosestNodes(infoHash);
			for(i, node in closestNodes) {

			dhtClient.sendGetPeers(node.ip, node.port,infoHash
				,onNodesOrPeersDiscovered); 
		}
		
	},1000
) 

// 定时发现新的 DHT 节点
winform.setInterval( 
	function(){
		var closestNodes = routingTable.getClosestNodes(string.random(20), 8);
		for(i, node in closestNodes) { 
			dhtClient.sendFindNode(node.ip, node.port, 
				string.random(20), onNodesOrPeersDiscovered);
		} 
	},1500 //当路由表增大发现新节点速度会越来越快，这时候应当增加延时，减慢速度
) 

// 收到 DHT 请求消息
dhtClient.onQuery = function(ip,port,nodeId){
	if( #nodeId == 20 ){
		
		var added = routingTable.addNode({
			ip = ip,
			port = port,
			nodeId = nodeId,
			lastSeen = time()
		});
		
		if(added){
			winform.editDhtNodes.print("添加新节点:", ip, port, "总节点数:", routingTable.getNodeCount());
		}
	}
}

// 定期清理旧节点
winform.setInterval(
	function(){
		routingTable.cleanOldNodes();

		winform.editDhtNodes.print("═════════ 路由表状态 ═════════");
		winform.editDhtNodes.print("总节点数:", routingTable.getNodeCount());
		winform.editDhtNodes.print("InfoHash 数:", infoHashQueue.count );
	},10000
)

winform.splitter.split(winform.editDhtNodes,winform.editMetadata);

if(_STUDIO_INVOKED){io.open(); }

winform.show(); 
win.loopMessage();

/*
可用 process.aria2 启动 BT 下载任务。

aria2 的本机 BT/DHT 服务端口可以通过本机 IP "127.0.0.1" 去连接。 
DHT 端口（process.aria2 对象的 dhtListeningPort 属性）可用 wsock.bt.metadataClient 连接。
BT 端口（process.aria2 对象的 listeningPort 属性）可用 wsock.bt.dhtClient 连接。
*/
```