如何从JS libp2p/kad-dht的QueryEvent提取数据？存储无结果求解

核心问题分析 #

1. 节点未启动：代码中注释了await node.start();，Libp2p节点未启动时DHT服务无法正常运行，put操作实际并未将数据写入网络。
2. DHT模式配置缺失：默认Kad-DHT以client模式运行，仅能查询数据，无法接收存储请求，需显式配置为server模式才能提供存储服务。
3. 数据转换函数未导入：提取代码中使用的uint8ArrayToString未明确导入，会导致运行时错误，无法正确解析数据。

修复后的存储代码 #

import { createLibp2p } from 'libp2p';
import { tcp } from '@libp2p/tcp';
import { noise } from '@chainsafe/libp2p-noise';
import { stdinToStream, streamToConsole } from './stream.js';
import { yamux } from '@chainsafe/libp2p-yamux';
import { kadDHT } from '@libp2p/kad-dht';
import { identify, identifyPush } from '@libp2p/identify';
import { fromString as uint8fromString } from 'uint8arrays/from-string';

const start_server = async () => {
    const node = await createLibp2p({
        addresses: {
            listen: ["/ip4/0.0.0.0/tcp/10333"]
        },
        transports: [tcp()],
        streamMuxers: [yamux()],
        connectionEncrypters: [noise()],
        services: {
            identify: identify(),
            // 配置DHT为server模式，允许存储数据
            dht: kadDHT({
                mode: 'server'
            })
        }
    });

    // 必须启动节点，否则DHT服务无法工作
    await node.start();

    node.addEventListener('peer:connect', (evt) => {
        console.log("New peer connected", evt.detail.remotePeer.toString());
    });

    await node.handle('/chat/1.0.0', async ({ stream }) => {
        console.log("New stream of chat protocol");
        stdinToStream(stream);
        streamToConsole(stream);

        try {
            // 存储数据并添加错误捕获
            await node.services.dht.put(uint8fromString('hello'), uint8fromString('world'));
            console.log('数据已成功存入DHT');

            // 在这里测试读取数据
            const queryResult = node.services.dht.get(uint8fromString('hello'));
            const value = await extractDHTValue(queryResult);
            console.log('读取到的DHT数据:', value);
        } catch (err) {
            console.error('DHT操作失败:', err);
        }
    });

    node.getMultiaddrs().forEach((ma) => {
        console.log('Listening on:', ma.toString());
    });

    console.log('Node started!');
};

start_server();

修复后的提取数据代码 #

// 导入必需的转换函数
import { toString as uint8ArrayToString } from 'uint8arrays/to-string';

const extractDHTValue = async (res) => {
    let finalValue = null;

    try {
        for await (const event of res) {
            // 处理VALUE事件，确保record和value存在
            if (event.name === 'VALUE' && event.record?.value) {
                const value = uint8ArrayToString(event.record.value);
                console.log('DHT Record Value:', value);
                finalValue = value;
                // 拿到第一个有效结果后可提前终止迭代
                break;
            }
            // 可选：处理其他事件用于调试
            if (event.name === 'PEER_RESPONSE') {
                console.log('收到节点响应:', event.peer.toString());
            }
        }
    } catch (err) {
        console.error('遍历QueryEvent失败:', err);
    }

    if (finalValue === null) {
        console.log('未在DHT查询结果中找到数据');
    }

    return finalValue;
};

额外注意事项 #

• 测试时确保网络中有至少一个节点（单节点环境可本地测试，生产环境需多节点保证DHT稳定性）。
• 多节点环境下需确认节点间已建立连接，可通过node.peerStore查看已连接节点列表。

内容的提问来源于stack exchange，提问作者Mathe Eliel