3
3
const net = require ( 'net' )
4
4
const EventEmitter = require ( 'events' )
5
5
const debug = require ( 'debug' )
6
- const log = debug ( 'libp2p:tcp:listener' )
7
- log . error = debug ( 'libp2p:tcp:listener:error' )
8
-
6
+ const log = Object . assign (
7
+ debug ( 'libp2p:tcp:listener' ) ,
8
+ { error : debug ( 'libp2p:tcp:listener:error' ) } )
9
9
const toConnection = require ( './socket-to-conn' )
10
10
const { CODE_P2P } = require ( './constants' )
11
11
const {
12
12
getMultiaddrs,
13
13
multiaddrToNetConfig
14
14
} = require ( './utils' )
15
15
16
+ /**
17
+ * @typedef {import('multiaddr').Multiaddr } Multiaddr
18
+ * @typedef {import('libp2p-interfaces/src/connection').Connection } Connection
19
+ * @typedef {import('libp2p-interfaces/src/transport/types').Upgrader } Upgrader
20
+ * @typedef {import('libp2p-interfaces/src/transport/types').MultiaddrConnection } MultiaddrConnection
21
+ * @typedef {import('libp2p-interfaces/src/transport/types').Listener } Listener
22
+ * @typedef {import('net').Server & {__connections: MultiaddrConnection[]} } Server
23
+ */
24
+
16
25
/**
17
26
* Attempts to close the given maConn. If a failure occurs, it will be logged.
18
27
*
19
28
* @private
20
- * @param {import('libp2p-interfaces/src/transport/types'). MultiaddrConnection } maConn
29
+ * @param {MultiaddrConnection } maConn
21
30
*/
22
31
async function attemptClose ( maConn ) {
23
32
try {
@@ -27,13 +36,80 @@ async function attemptClose (maConn) {
27
36
}
28
37
}
29
38
39
+ /**
40
+ * Create listener
41
+ *
42
+ * @param {object } context
43
+ * @param {function(Connection):void } context.handler
44
+ * @param {Upgrader } context.upgrader
45
+ * @param {* } options
46
+ * @returns {Listener }
47
+ */
30
48
module . exports = ( { handler, upgrader } , options ) => {
31
- const listener = new EventEmitter ( )
49
+ /** @type {Server } */
50
+ // eslint-disable-next-line prefer-const
51
+ let server
52
+
53
+ /** @type {string | null } */
54
+ let peerId
55
+
56
+ /** @type {Multiaddr } */
57
+ let listeningAddr
58
+
59
+ const listener = Object . assign ( new EventEmitter ( ) , {
60
+ getAddrs : ( ) => {
61
+ /** @type {Multiaddr[] } */
62
+ let addrs = [ ]
63
+ /** @type {import('net').AddressInfo } */
64
+ // @ts -ignore
65
+ const address = server . address ( )
66
+
67
+ if ( ! address ) {
68
+ throw new Error ( 'Listener is not ready yet' )
69
+ }
70
+
71
+ // Because TCP will only return the IPv6 version
72
+ // we need to capture from the passed multiaddr
73
+ if ( listeningAddr . toString ( ) . startsWith ( '/ip4' ) ) {
74
+ addrs = addrs . concat ( getMultiaddrs ( 'ip4' , address . address , address . port ) )
75
+ } else if ( address . family === 'IPv6' ) {
76
+ addrs = addrs . concat ( getMultiaddrs ( 'ip6' , address . address , address . port ) )
77
+ }
78
+
79
+ return addrs . map ( ma => peerId ? ma . encapsulate ( `/p2p/${ peerId } ` ) : ma )
80
+ } ,
81
+ listen : async ( /** @type {Multiaddr } */ ma ) => {
82
+ listeningAddr = ma
83
+ peerId = ma . getPeerId ( )
84
+
85
+ if ( peerId ) {
86
+ listeningAddr = ma . decapsulateCode ( CODE_P2P )
87
+ }
88
+
89
+ return new Promise ( ( resolve , reject ) => {
90
+ const options = multiaddrToNetConfig ( listeningAddr )
91
+ server . listen ( options , ( /** @type {any } */ err ) => {
92
+ if ( err ) return reject ( err )
93
+ log ( 'Listening on %s' , server . address ( ) )
94
+ resolve ( undefined )
95
+ } )
96
+ } )
97
+ } ,
98
+ close : async ( ) => {
99
+ if ( ! server . listening ) return
100
+
101
+ return new Promise ( ( resolve , reject ) => {
102
+ server . __connections . forEach ( maConn => attemptClose ( maConn ) )
103
+ server . close ( err => err ? reject ( err ) : resolve ( undefined ) )
104
+ } )
105
+ }
106
+ } )
32
107
33
- const server = net . createServer ( async socket => {
108
+ server = Object . assign ( net . createServer ( async socket => {
34
109
// Avoid uncaught errors caused by unstable connections
35
110
socket . on ( 'error' , err => log ( 'socket error' , err ) )
36
111
112
+ /** @type {MultiaddrConnection } */
37
113
let maConn
38
114
let conn
39
115
try {
@@ -42,6 +118,7 @@ module.exports = ({ handler, upgrader }, options) => {
42
118
conn = await upgrader . upgradeInbound ( maConn )
43
119
} catch ( err ) {
44
120
log . error ( 'inbound connection failed' , err )
121
+ // @ts -ignore
45
122
return attemptClose ( maConn )
46
123
}
47
124
@@ -51,73 +128,29 @@ module.exports = ({ handler, upgrader }, options) => {
51
128
52
129
if ( handler ) handler ( conn )
53
130
listener . emit ( 'connection' , conn )
54
- } )
131
+ } ) ,
132
+ // Keep track of open connections to destroy in case of timeout
133
+ { __connections : [ ] } )
55
134
56
135
server
57
136
. on ( 'listening' , ( ) => listener . emit ( 'listening' ) )
58
137
. on ( 'error' , err => listener . emit ( 'error' , err ) )
59
138
. on ( 'close' , ( ) => listener . emit ( 'close' ) )
60
139
61
- // Keep track of open connections to destroy in case of timeout
62
- server . __connections = [ ]
63
-
64
- listener . close = ( ) => {
65
- if ( ! server . listening ) return
66
-
67
- return new Promise ( ( resolve , reject ) => {
68
- server . __connections . forEach ( maConn => attemptClose ( maConn ) )
69
- server . close ( err => err ? reject ( err ) : resolve ( ) )
70
- } )
71
- }
72
-
73
- let peerId , listeningAddr
74
-
75
- listener . listen = ma => {
76
- listeningAddr = ma
77
- peerId = ma . getPeerId ( )
78
-
79
- if ( peerId ) {
80
- listeningAddr = ma . decapsulateCode ( CODE_P2P )
81
- }
82
-
83
- return new Promise ( ( resolve , reject ) => {
84
- const options = multiaddrToNetConfig ( listeningAddr )
85
- server . listen ( options , err => {
86
- if ( err ) return reject ( err )
87
- log ( 'Listening on %s' , server . address ( ) )
88
- resolve ( )
89
- } )
90
- } )
91
- }
92
-
93
- listener . getAddrs = ( ) => {
94
- let addrs = [ ]
95
- const address = server . address ( )
96
-
97
- if ( ! address ) {
98
- throw new Error ( 'Listener is not ready yet' )
99
- }
100
-
101
- // Because TCP will only return the IPv6 version
102
- // we need to capture from the passed multiaddr
103
- if ( listeningAddr . toString ( ) . startsWith ( '/ip4' ) ) {
104
- addrs = addrs . concat ( getMultiaddrs ( 'ip4' , address . address , address . port ) )
105
- } else if ( address . family === 'IPv6' ) {
106
- addrs = addrs . concat ( getMultiaddrs ( 'ip6' , address . address , address . port ) )
107
- }
108
-
109
- return addrs . map ( ma => peerId ? ma . encapsulate ( `/p2p/${ peerId } ` ) : ma )
110
- }
111
-
112
140
return listener
113
141
}
114
142
143
+ /**
144
+ * @param {Server } server
145
+ * @param {MultiaddrConnection } maConn
146
+ */
115
147
function trackConn ( server , maConn ) {
116
148
server . __connections . push ( maConn )
117
149
118
150
const untrackConn = ( ) => {
119
151
server . __connections = server . __connections . filter ( c => c !== maConn )
120
152
}
121
153
154
+ // @ts -ignore
122
155
maConn . conn . once ( 'close' , untrackConn )
123
156
}
0 commit comments