Skip to content
This repository was archived by the owner on Jul 21, 2023. It is now read-only.

Commit 388042b

Browse files
achingbrainachingbrain
authored
chore: update interfaces to new version (#327)
Removes dialer, lets connection manager manage connections.
1 parent 526e65e commit 388042b

16 files changed

+140
-186
lines changed

‎.aegir.js

+1-1
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,6 @@
22
/** @type {import('aegir').PartialOptions} */
33
export default {
44
build: {
5-
bundlesizeMax: '300KB'
5+
bundlesizeMax: '160KB'
66
}
77
}

‎package.json

+8-7
Original file line numberDiff line numberDiff line change
@@ -132,13 +132,14 @@
132132
"release": "aegir release"
133133
},
134134
"dependencies": {
135-
"@libp2p/crypto": "^0.22.10",
136-
"@libp2p/interfaces": "^1.3.21",
137-
"@libp2p/logger": "^1.1.3",
138-
"@libp2p/peer-id": "^1.1.9",
139-
"@libp2p/record": "^1.0.3",
135+
"@libp2p/crypto": "^0.22.11",
136+
"@libp2p/interfaces": "^1.3.29",
137+
"@libp2p/logger": "^1.1.4",
138+
"@libp2p/peer-id": "^1.1.10",
139+
"@libp2p/record": "^1.0.4",
140140
"@libp2p/topology": "^1.1.7",
141141
"@multiformats/multiaddr": "^10.1.5",
142+
"abortable-iterator": "^4.0.2",
142143
"any-signal": "^3.0.0",
143144
"datastore-core": "^7.0.0",
144145
"err-code": "^3.0.1",
@@ -168,9 +169,9 @@
168169
"varint": "^6.0.0"
169170
},
170171
"devDependencies": {
171-
"@libp2p/interface-compliance-tests": "^1.1.21",
172+
"@libp2p/interface-compliance-tests": "^1.1.31",
172173
"@libp2p/peer-id-factory": "^1.0.9",
173-
"@libp2p/peer-store": "^1.0.8",
174+
"@libp2p/peer-store": "^1.0.11",
174175
"@types/lodash.random": "^3.2.6",
175176
"@types/lodash.range": "^3.2.6",
176177
"@types/node": "^16.11.26",

‎src/content-fetching/index.ts

+1
Original file line numberDiff line numberDiff line change
@@ -163,6 +163,7 @@ export class ContentFetching implements Initializable {
163163
const msg = new Message(MESSAGE_TYPE.PUT_VALUE, key, 0)
164164
msg.record = Libp2pRecord.deserialize(record)
165165

166+
this.log('send put to %p', event.peer.id)
166167
for await (const putEvent of this.network.sendRequest(event.peer.id, msg, options)) {
167168
events.push(putEvent)
168169

‎src/kad-dht.ts

+1-2
Original file line numberDiff line numberDiff line change
@@ -196,7 +196,7 @@ export class KadDHT extends EventEmitter<PeerDiscoveryEvents> implements DHT, In
196196
}
197197

198198
async onPeerConnect (peerData: PeerInfo) {
199-
this.log('peer %p connected', peerData.id)
199+
this.log('peer %p connected with protocols %s', peerData.id, peerData.protocols)
200200

201201
if (this.lan) {
202202
peerData = removePublicAddresses(peerData)
@@ -242,7 +242,6 @@ export class KadDHT extends EventEmitter<PeerDiscoveryEvents> implements DHT, In
242242
} else {
243243
this.log('enabling server mode')
244244
this.clientMode = false
245-
246245
await this.components.getRegistrar().handle(this.protocol, this.rpc.onIncomingStream.bind(this.rpc))
247246
}
248247
}

‎src/network.ts

+35-13
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,8 @@ import type { Logger } from '@libp2p/logger'
1818
import type { Duplex } from 'it-stream-types'
1919
import type { PeerInfo } from '@libp2p/interfaces/peer-info'
2020
import { Components, Initializable } from '@libp2p/interfaces/components'
21+
import type { Stream } from '@libp2p/interfaces/connection'
22+
import { abortableDuplex } from 'abortable-iterator'
2123

2224
export interface NetworkInit {
2325
protocol: string
@@ -87,15 +89,17 @@ export class Network extends EventEmitter<NetworkEvents> implements Startable, I
8789
}
8890

8991
this.log('sending %s to %p', msg.type, to)
92+
yield dialingPeerEvent({ peer: to })
93+
yield sendingQueryEvent({ to, type: msg.type })
9094

91-
try {
92-
yield dialingPeerEvent({ peer: to })
93-
94-
const { stream } = await this.components.getDialer().dialProtocol(to, this.protocol, options)
95+
let stream: Stream | undefined
9596

96-
yield sendingQueryEvent({ to, type: msg.type })
97+
try {
98+
const connection = await this.components.getConnectionManager().openConnection(to, options)
99+
const streamData = await connection.newStream(this.protocol)
100+
stream = streamData.stream
97101

98-
const response = await this._writeReadMessage(stream, msg.serialize())
102+
const response = await this._writeReadMessage(stream, msg.serialize(), options)
99103

100104
yield peerResponseEvent({
101105
from: to,
@@ -106,6 +110,10 @@ export class Network extends EventEmitter<NetworkEvents> implements Startable, I
106110
})
107111
} catch (err: any) {
108112
yield queryErrorEvent({ from: to, error: err })
113+
} finally {
114+
if (stream != null) {
115+
stream.close()
116+
}
109117
}
110118
}
111119

@@ -118,26 +126,36 @@ export class Network extends EventEmitter<NetworkEvents> implements Startable, I
118126
}
119127

120128
this.log('sending %s to %p', msg.type, to)
121-
122129
yield dialingPeerEvent({ peer: to })
123-
124-
const { stream } = await this.components.getDialer().dialProtocol(to, this.protocol, options)
125-
126130
yield sendingQueryEvent({ to, type: msg.type })
127131

132+
let stream: Stream | undefined
133+
128134
try {
129-
await this._writeMessage(stream, msg.serialize())
135+
const connection = await this.components.getConnectionManager().openConnection(to, options)
136+
const data = await connection.newStream(this.protocol)
137+
stream = data.stream
138+
139+
await this._writeMessage(stream, msg.serialize(), options)
130140

131141
yield peerResponseEvent({ from: to, messageType: msg.type })
132142
} catch (err: any) {
133143
yield queryErrorEvent({ from: to, error: err })
144+
} finally {
145+
if (stream != null) {
146+
stream.close()
147+
}
134148
}
135149
}
136150

137151
/**
138152
* Write a message to the given stream
139153
*/
140-
async _writeMessage (stream: Duplex<Uint8Array>, msg: Uint8Array) {
154+
async _writeMessage (stream: Duplex<Uint8Array>, msg: Uint8Array, options: AbortOptions) {
155+
if (options.signal != null) {
156+
stream = abortableDuplex(stream, options.signal)
157+
}
158+
141159
await pipe(
142160
[msg],
143161
lp.encode(),
@@ -151,7 +169,11 @@ export class Network extends EventEmitter<NetworkEvents> implements Startable, I
151169
* If no response is received after the specified timeout
152170
* this will error out.
153171
*/
154-
async _writeReadMessage (stream: Duplex<Uint8Array>, msg: Uint8Array) {
172+
async _writeReadMessage (stream: Duplex<Uint8Array>, msg: Uint8Array, options: AbortOptions) {
173+
if (options.signal != null) {
174+
stream = abortableDuplex(stream, options.signal)
175+
}
176+
155177
const res = await pipe(
156178
[msg],
157179
lp.encode(),

‎src/routing-table/index.ts

+2-1
Original file line numberDiff line numberDiff line change
@@ -128,9 +128,10 @@ export class RoutingTable implements Startable, Initializable {
128128
timeoutController = new TimeoutController(this.pingTimeout)
129129

130130
this.log('pinging old contact %p', oldContact.peer)
131-
const { stream } = await this.components.getDialer().dialProtocol(oldContact.peer, PROTOCOL_DHT, {
131+
const connection = await this.components.getConnectionManager().openConnection(oldContact.peer, {
132132
signal: timeoutController.signal
133133
})
134+
const { stream } = await connection.newStream(PROTOCOL_DHT)
134135
await stream.close()
135136
responded++
136137
} catch (err: any) {

‎src/rpc/index.ts

+1-1
Original file line numberDiff line numberDiff line change
@@ -98,7 +98,7 @@ export class RPC implements Initializable {
9898
source => (async function * () {
9999
for await (const msg of source) {
100100
// handle the message
101-
const desMessage = Message.deserialize(msg.slice())
101+
const desMessage = Message.deserialize(msg)
102102
self.log('incoming %s from %p', desMessage.type, peerId)
103103
const res = await self.handleMessage(peerId, desMessage)
104104

‎src/topology-listener.ts

+1-1
Original file line numberDiff line numberDiff line change
@@ -56,7 +56,7 @@ export class TopologyListener extends EventEmitter<TopologyListenerEvents> imple
5656
// register protocol with topology
5757
const topology = createTopology({
5858
onConnect: (peerId) => {
59-
this.log('observed peer %p with protocol %s', this.protocol, peerId)
59+
this.log('observed peer %p with protocol %s', peerId, this.protocol)
6060
this.dispatchEvent(new CustomEvent('peer', {
6161
detail: peerId
6262
}))

‎test/generate-peers/generate-peers.spec.ts

+2-2
Original file line numberDiff line numberDiff line change
@@ -11,9 +11,9 @@ import {
1111
} from '../../src/utils.js'
1212
import { Components } from '@libp2p/interfaces/components'
1313
import { stubInterface } from 'ts-sinon'
14-
import type { Dialer } from '@libp2p/interfaces/dialer'
1514
import path from 'path'
1615
import { fileURLToPath } from 'url'
16+
import type { ConnectionManager } from '@libp2p/interfaces/connection-manager'
1717

1818
const dirname = path.dirname(fileURLToPath(import.meta.url))
1919

@@ -54,7 +54,7 @@ describe.skip('generate peers', function () {
5454

5555
const components = new Components({
5656
peerId: id,
57-
dialer: stubInterface<Dialer>()
57+
connectionManager: stubInterface<ConnectionManager>()
5858
})
5959
const table = new RoutingTable({
6060
kBucketSize: 20,

‎test/kad-dht.spec.ts

+9-5
Original file line numberDiff line numberDiff line change
@@ -750,10 +750,14 @@ describe('KadDHT', () => {
750750
new Array(nDHTs).fill(0).map(async () => await tdht.spawn())
751751
)
752752

753+
const connected: Array<Promise<void>> = []
754+
753755
for (let i = 0; i < dhts.length - 1; i++) {
754-
await tdht.connect(dhts[i], dhts[(i + 1) % dhts.length])
756+
connected.push(tdht.connect(dhts[i], dhts[(i + 1) % dhts.length]))
755757
}
756758

759+
await Promise.all(connected)
760+
757761
const res = await all(filter(dhts[1].getClosestPeers(uint8ArrayFromString('foo')), event => event.name === 'FINAL_PEER'))
758762

759763
expect(res).to.not.be.empty()
@@ -785,16 +789,16 @@ describe('KadDHT', () => {
785789
tdht.spawn(),
786790
tdht.spawn()
787791
])
788-
const stub = sinon.stub(dhtA.components.getDialer(), 'dialProtocol').rejects(error)
789792

790793
await tdht.connect(dhtA, dhtB)
791794

795+
const stub = sinon.stub(dhtA.components.getConnectionManager(), 'openConnection').rejects(error)
796+
792797
const errors = await all(filter(dhtA.get(uint8ArrayFromString('/v/hello')), event => event.name === 'QUERY_ERROR'))
793798

794-
expect(errors).to.have.lengthOf(3)
799+
expect(errors).to.have.lengthOf(2)
795800
expect(errors).to.have.nested.property('[0].error.code', errCode)
796-
expect(errors).to.have.nested.property('[1].error.code', errCode)
797-
expect(errors).to.have.nested.property('[2].error.code', 'ERR_NOT_FOUND')
801+
expect(errors).to.have.nested.property('[1].error.code', 'ERR_NOT_FOUND')
798802

799803
stub.restore()
800804
})

‎test/network.spec.ts

+37-42
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,6 @@
11
/* eslint-env mocha */
22

33
import { expect } from 'aegir/chai'
4-
import { pair } from 'it-pair'
54
import { pipe } from 'it-pipe'
65
import * as lp from 'it-length-prefixed'
76
import pDefer from 'p-defer'
@@ -11,9 +10,9 @@ import { Message, MESSAGE_TYPE } from '../src/message/index.js'
1110
import { TestDHT } from './utils/test-dht.js'
1211
import { mockStream } from '@libp2p/interface-compliance-tests/mocks'
1312
import type { DualKadDHT } from '../src/dual-kad-dht.js'
14-
import type { Sink } from 'it-stream-types'
15-
import type { Multiaddr } from '@multiformats/multiaddr'
13+
import type { Connection } from '@libp2p/interfaces/connection'
1614
import type { PeerId } from '@libp2p/interfaces/peer-id'
15+
import type { Sink } from 'it-stream-types'
1716

1817
describe('Network', () => {
1918
let dht: DualKadDHT
@@ -33,17 +32,6 @@ describe('Network', () => {
3332
it('send and response echo', async () => {
3433
const msg = new Message(MESSAGE_TYPE.PING, uint8ArrayFromString('hello'), 0)
3534

36-
// mock dial
37-
dht.components.getDialer().dialProtocol = async (peer: PeerId | Multiaddr, protocols: string | string[]) => {
38-
const protocol = Array.isArray(protocols) ? protocols[0] : protocols
39-
40-
// {source, sink} streams that are internally connected
41-
return {
42-
stream: mockStream(pair()),
43-
protocol
44-
}
45-
}
46-
4735
const events = await all(dht.lan.network.sendRequest(dht.components.getPeerId(), msg))
4836
const response = events
4937
.filter(event => event.name === 'PEER_RESPONSE')
@@ -63,36 +51,43 @@ describe('Network', () => {
6351
const msg = new Message(MESSAGE_TYPE.PING, uint8ArrayFromString('hello'), 0)
6452

6553
// mock it
66-
dht.components.getDialer().dialProtocol = async (peer: PeerId | Multiaddr, protocols: string | string[]) => {
67-
const protocol = Array.isArray(protocols) ? protocols[0] : protocols
68-
const msg = new Message(MESSAGE_TYPE.FIND_NODE, uint8ArrayFromString('world'), 0)
69-
70-
const data = await pipe(
71-
[msg.serialize()],
72-
lp.encode(),
73-
async (source) => await all(source)
74-
)
75-
76-
const source = (function * () {
77-
const array = data
78-
79-
yield * array
80-
})()
81-
82-
const sink: Sink<Uint8Array> = async source => {
83-
const res = await pipe(
84-
source,
85-
lp.decode(),
86-
async (source) => await all(source)
87-
)
88-
expect(Message.deserialize(res[0]).type).to.eql(MESSAGE_TYPE.PING)
89-
finish()
54+
dht.components.getConnectionManager().openConnection = async (peer: PeerId) => {
55+
// @ts-expect-error incomplete implementation
56+
const connection: Connection = {
57+
newStream: async (protocols: string | string[]) => {
58+
const protocol = Array.isArray(protocols) ? protocols[0] : protocols
59+
const msg = new Message(MESSAGE_TYPE.FIND_NODE, uint8ArrayFromString('world'), 0)
60+
61+
const data = await pipe(
62+
[msg.serialize()],
63+
lp.encode(),
64+
async (source) => await all(source)
65+
)
66+
67+
const source = (function * () {
68+
const array = data
69+
70+
yield * array
71+
})()
72+
73+
const sink: Sink<Uint8Array> = async source => {
74+
const res = await pipe(
75+
source,
76+
lp.decode(),
77+
async (source) => await all(source)
78+
)
79+
expect(Message.deserialize(res[0]).type).to.eql(MESSAGE_TYPE.PING)
80+
finish()
81+
}
82+
83+
return {
84+
protocol,
85+
stream: mockStream({ source, sink })
86+
}
87+
}
9088
}
9189

92-
return {
93-
protocol,
94-
stream: mockStream({ source, sink })
95-
}
90+
return connection
9691
}
9792

9893
const events = await all(dht.lan.network.sendRequest(dht.components.getPeerId(), msg))

0 commit comments

Comments
 (0)