From 37adc28ec6edd7db08166984eda4da2e56ac7ba3 Mon Sep 17 00:00:00 2001 From: Alex Potsides Date: Fri, 3 Dec 2021 10:45:12 +0000 Subject: [PATCH] feat: dht client (#3947) * Enables `libp2p-kad-dht` in client mode * Updates types with new DHT events BREAKING CHANGE: The DHT API has been refactored to return async iterators of query events --- src/dht/find-peer.js | 17 ++---- src/dht/find-provs.js | 14 +---- src/dht/get.js | 16 ++--- src/dht/map-event.js | 119 ++++++++++++++++++++++++++++++++++++++ src/dht/provide.js | 16 +---- src/dht/put.js | 16 ++--- src/dht/query.js | 12 +--- src/dht/response-types.js | 2 +- 8 files changed, 143 insertions(+), 69 deletions(-) create mode 100644 src/dht/map-event.js diff --git a/src/dht/find-peer.js b/src/dht/find-peer.js index 7077095f6..22c76b563 100644 --- a/src/dht/find-peer.js +++ b/src/dht/find-peer.js @@ -1,7 +1,6 @@ -import { Multiaddr } from 'multiaddr' import { configure } from '../lib/configure.js' import { toUrlSearchParams } from '../lib/to-url-search-params.js' -import { FinalPeer } from './response-types.js' +import { mapEvent } from './map-event.js' /** * @typedef {import('../types').HTTPClientExtraOptions} HTTPClientExtraOptions @@ -12,7 +11,7 @@ export const createFindPeer = configure(api => { /** * @type {DHTAPI["findPeer"]} */ - async function findPeer (peerId, options = {}) { + async function * findPeer (peerId, options = {}) { const res = await api.post('dht/findpeer', { signal: options.signal, searchParams: toUrlSearchParams({ @@ -22,17 +21,9 @@ export const createFindPeer = configure(api => { headers: options.headers }) - for await (const data of res.ndjson()) { - if (data.Type === FinalPeer && data.Responses) { - const { ID, Addrs } = data.Responses[0] - return { - id: ID, - addrs: (Addrs || []).map((/** @type {string} **/ a) => new Multiaddr(a)) - } - } + for await (const event of res.ndjson()) { + yield mapEvent(event) } - - throw new Error('not found') } return findPeer diff --git a/src/dht/find-provs.js b/src/dht/find-provs.js index 915d511ea..46f0312cd 100644 --- a/src/dht/find-provs.js +++ b/src/dht/find-provs.js @@ -1,7 +1,6 @@ -import { Multiaddr } from 'multiaddr' import { configure } from '../lib/configure.js' import { toUrlSearchParams } from '../lib/to-url-search-params.js' -import { Provider } from './response-types.js' +import { mapEvent } from './map-event.js' /** * @typedef {import('../types').HTTPClientExtraOptions} HTTPClientExtraOptions @@ -22,15 +21,8 @@ export const createFindProvs = configure(api => { headers: options.headers }) - for await (const message of res.ndjson()) { - if (message.Type === Provider && message.Responses) { - for (const { ID, Addrs } of message.Responses) { - yield { - id: ID, - addrs: (Addrs || []).map((/** @type {string} **/ a) => new Multiaddr(a)) - } - } - } + for await (const event of res.ndjson()) { + yield mapEvent(event) } } diff --git a/src/dht/get.js b/src/dht/get.js index 651978692..db9dac593 100644 --- a/src/dht/get.js +++ b/src/dht/get.js @@ -1,7 +1,6 @@ import { configure } from '../lib/configure.js' import { toUrlSearchParams } from '../lib/to-url-search-params.js' -import { Value } from './response-types.js' -import { fromString as uint8ArrayFromString } from 'uint8arrays/from-string' +import { mapEvent } from './map-event.js' import { toString as uint8ArrayToString } from 'uint8arrays/to-string' /** @@ -13,23 +12,20 @@ export const createGet = configure(api => { /** * @type {DHTAPI["get"]} */ - async function get (key, options = {}) { + async function * get (key, options = {}) { const res = await api.post('dht/get', { signal: options.signal, searchParams: toUrlSearchParams({ - arg: key instanceof Uint8Array ? uint8ArrayToString(key) : key, + // arg: base36.encode(key), + arg: key instanceof Uint8Array ? uint8ArrayToString(key) : key.toString(), ...options }), headers: options.headers }) - for await (const message of res.ndjson()) { - if (message.Type === Value) { - return uint8ArrayFromString(message.Extra, 'base64pad') - } + for await (const event of res.ndjson()) { + yield mapEvent(event) } - - throw new Error('not found') } return get diff --git a/src/dht/map-event.js b/src/dht/map-event.js new file mode 100644 index 000000000..af641d042 --- /dev/null +++ b/src/dht/map-event.js @@ -0,0 +1,119 @@ +import { fromString as uint8ArrayFromString } from 'uint8arrays/from-string' +import { + SendingQuery, + PeerResponse, + FinalPeer, + QueryError, + Provider, + Value, + AddingPeer, + DialingPeer +} from './response-types.js' +import { Multiaddr } from 'multiaddr' + +/** + * @param {{Type: number, ID: string, Extra: string, Responses: {ID: string, Addrs: string[]}[]}} event + * @returns {import('ipfs-core-types/src/dht').QueryEvent} + */ +export const mapEvent = (event) => { + // console.info(JSON.stringify(event, null, 2)) + + if (event.Type === SendingQuery) { + return { + to: event.ID, + name: 'SENDING_QUERY', + type: event.Type + } + } + + if (event.Type === PeerResponse) { + return { + from: event.ID, + name: 'PEER_RESPONSE', + type: event.Type, + // TODO: how to infer this from the go-ipfs response + messageType: 0, + // TODO: how to infer this from the go-ipfs response + messageName: 'PUT_VALUE', + closer: (event.Responses || []).map(({ ID, Addrs }) => ({ id: ID, multiaddrs: Addrs.map(addr => new Multiaddr(addr)) })), + providers: (event.Responses || []).map(({ ID, Addrs }) => ({ id: ID, multiaddrs: Addrs.map(addr => new Multiaddr(addr)) })) + // TODO: how to infer this from the go-ipfs response + // record: ??? + } + } + + if (event.Type === FinalPeer) { + // dht.query ends with a FinalPeer event with no Responses + let peer = { + id: event.ID, + /** @type {Multiaddr[]} */ + multiaddrs: [] + } + + if (event.Responses && event.Responses.length) { + // dht.findPeer has the result in the Responses field + peer = { + id: event.Responses[0].ID, + multiaddrs: event.Responses[0].Addrs.map(addr => new Multiaddr(addr)) + } + } + + return { + from: event.ID, + name: 'FINAL_PEER', + type: event.Type, + peer + } + } + + if (event.Type === QueryError) { + return { + from: event.ID, + name: 'QUERY_ERROR', + type: event.Type, + error: new Error(event.Extra) + } + } + + if (event.Type === Provider) { + return { + from: event.ID, + name: 'PROVIDER', + type: event.Type, + providers: event.Responses.map(({ ID, Addrs }) => ({ id: ID, multiaddrs: Addrs.map(addr => new Multiaddr(addr)) })) + } + } + + if (event.Type === Value) { + return { + from: event.ID, + name: 'VALUE', + type: event.Type, + value: uint8ArrayFromString(event.Extra, 'base64pad') + } + } + + if (event.Type === AddingPeer) { + const peers = event.Responses.map(({ ID }) => ID) + + if (!peers.length) { + throw new Error('No peer found') + } + + return { + name: 'ADDING_PEER', + type: event.Type, + peer: peers[0] + } + } + + if (event.Type === DialingPeer) { + return { + name: 'DIALING_PEER', + type: event.Type, + peer: event.ID + } + } + + throw new Error('Unknown DHT event type') +} diff --git a/src/dht/provide.js b/src/dht/provide.js index 71b23e503..baeefbb2c 100644 --- a/src/dht/provide.js +++ b/src/dht/provide.js @@ -1,7 +1,6 @@ -import { Multiaddr } from 'multiaddr' -import { objectToCamel } from '../lib/object-to-camel.js' import { configure } from '../lib/configure.js' import { toUrlSearchParams } from '../lib/to-url-search-params.js' +import { mapEvent } from './map-event.js' /** * @typedef {import('../types').HTTPClientExtraOptions} HTTPClientExtraOptions @@ -26,17 +25,8 @@ export const createProvide = configure(api => { headers: options.headers }) - for await (let message of res.ndjson()) { - message = objectToCamel(message) - if (message.responses) { - message.responses = message.responses.map((/** @type {{ ID: string, Addrs: string[] }} */ { ID, Addrs }) => ({ - id: ID, - addrs: (Addrs || []).map((/** @type {string} **/ a) => new Multiaddr(a)) - })) - } else { - message.responses = [] - } - yield message + for await (const event of res.ndjson()) { + yield mapEvent(event) } } diff --git a/src/dht/put.js b/src/dht/put.js index 259a70884..807899df3 100644 --- a/src/dht/put.js +++ b/src/dht/put.js @@ -1,11 +1,10 @@ -import { Multiaddr } from 'multiaddr' -import { objectToCamel } from '../lib/object-to-camel.js' import { configure } from '../lib/configure.js' import { toUrlSearchParams } from '../lib/to-url-search-params.js' import { multipartRequest } from 'ipfs-core-utils/multipart-request' import { abortSignal } from '../lib/abort-signal.js' import { AbortController } from 'native-abort-controller' import { toString as uint8ArrayToString } from 'uint8arrays/to-string' +import { mapEvent } from './map-event.js' /** * @typedef {import('../types').HTTPClientExtraOptions} HTTPClientExtraOptions @@ -24,7 +23,7 @@ export const createPut = configure(api => { const res = await api.post('dht/put', { signal, searchParams: toUrlSearchParams({ - arg: uint8ArrayToString(key), + arg: key instanceof Uint8Array ? uint8ArrayToString(key) : key.toString(), ...options }), ...( @@ -32,15 +31,8 @@ export const createPut = configure(api => { ) }) - for await (let message of res.ndjson()) { - message = objectToCamel(message) - if (message.responses) { - message.responses = message.responses.map((/** @type {{ ID: string, Addrs: string[] }} */ { ID, Addrs }) => ({ - id: ID, - addrs: (Addrs || []).map(a => new Multiaddr(a)) - })) - } - yield message + for await (const event of res.ndjson()) { + yield mapEvent(event) } } diff --git a/src/dht/query.js b/src/dht/query.js index 0923f44bb..3f4465eae 100644 --- a/src/dht/query.js +++ b/src/dht/query.js @@ -1,7 +1,6 @@ -import { Multiaddr } from 'multiaddr' -import { objectToCamel } from '../lib/object-to-camel.js' import { configure } from '../lib/configure.js' import { toUrlSearchParams } from '../lib/to-url-search-params.js' +import { mapEvent } from './map-event.js' /** * @typedef {import('../types').HTTPClientExtraOptions} HTTPClientExtraOptions @@ -22,13 +21,8 @@ export const createQuery = configure(api => { headers: options.headers }) - for await (let message of res.ndjson()) { - message = objectToCamel(message) - message.responses = (message.responses || []).map((/** @type {{ ID: string, Addrs: string[] }} */ { ID, Addrs }) => ({ - id: ID, - addrs: (Addrs || []).map((/** @type {string} **/ a) => new Multiaddr(a)) - })) - yield message + for await (const event of res.ndjson()) { + yield mapEvent(event) } } diff --git a/src/dht/response-types.js b/src/dht/response-types.js index 24250e4a1..a30e846cb 100644 --- a/src/dht/response-types.js +++ b/src/dht/response-types.js @@ -1,6 +1,6 @@ // Response types are defined here = -// https =//github.com/libp2p/go-libp2p-core/blob/6e566d10f4a5447317a66d64c7459954b969bdab/routing/query.go#L15-L24 +// https://github.com/libp2p/go-libp2p-core/blob/6e566d10f4a5447317a66d64c7459954b969bdab/routing/query.go#L15-L24 export const SendingQuery = 0 export const PeerResponse = 1 export const FinalPeer = 2