You signed in with another tab or window. Reload to refresh your session.You signed out in another tab or window. Reload to refresh your session.You switched accounts on another tab or window. Reload to refresh your session.Dismiss alert
{{ message }}
This repository was archived by the owner on Jul 21, 2023. It is now read-only.
BREAKING CHANGE: All places in the API that used callbacks are now replaced with async/await while pull-streams are replaced with async iterators. The API has also been updated according to the latest `interface-stream-muxer` version, https://github.com/libp2p/interface-stream-muxer/tree/v0.7.0.
License: MIT
Signed-off-by: Alan Shaw <alan.shaw@protocol.ai>
Let's define a `listener.js`, which starts a TCP server on port 9999 and waits for a connection. Once we get a connection, we wait for a stream. And finally, once we have the stream, we pull the data from that stream, and printing it to the console.
26
-
27
-
```JavaScript
28
-
constmplex=require('libp2p-mplex')
29
-
consttcp=require('net')
30
-
constpull=require('pull-stream')
31
-
consttoPull=require('stream-to-pull-stream')
32
-
33
-
constlistener=tcp.createServer((socket) => {
34
-
console.log('[listener] Got connection!')
35
-
36
-
constmuxer=mplex.listener(toPull(socket))
37
-
38
-
muxer.on('stream', (stream) => {
39
-
console.log('[listener] Got stream!')
40
-
pull(
41
-
stream,
42
-
pull.drain((data) => {
43
-
console.log('[listener] Received:')
44
-
console.log(data.toString())
45
-
})
46
-
)
47
-
})
48
-
})
30
+
```js
31
+
constMplex=require('libp2p-mplex')
32
+
constpipe=require('it-pipe')
49
33
50
-
listener.listen(9999, () => {
51
-
console.log('[listener] listening on 9999')
34
+
constmuxer=newMplex({
35
+
onStream:stream=> { // Receive a duplex stream from the remote
36
+
// ...receive data from the remote and optionally send data back
37
+
}
52
38
})
39
+
40
+
pipe(conn, muxer, conn) // conn is duplex connection to another peer
41
+
42
+
conststream=muxer.newStream() // Create a new duplex stream to the remote
43
+
44
+
// Use the duplex stream to send some data to the remote...
45
+
pipe([1, 2, 3], stream)
53
46
```
54
47
55
-
Now, let's define `dialer.js` who will connect to our `listener` over a TCP socket. Once we have that, we'll put a message in the stream for our `listener`.
48
+
## API
56
49
57
-
```JavaScript
58
-
constmplex=require('libp2p-mplex')
59
-
consttcp=require('net')
60
-
constpull=require('pull-stream')
61
-
consttoPull=require('stream-to-pull-stream')
50
+
### `const muxer = new Mplex([options])`
62
51
63
-
constsocket=tcp.connect(9999)
52
+
Create a new _duplex_ stream that can be piped together with a connection in order to allow multiplexed communications.
64
53
65
-
constmuxer=mplex.dialer(toPull(socket))
54
+
e.g.
66
55
67
-
console.log('[dialer] opening stream')
68
-
conststream=muxer.newStream((err) => {
69
-
console.log('[dialer] opened stream')
70
-
if (err) throw err
71
-
})
56
+
```js
57
+
constMplex=require('libp2p-mplex')
58
+
constpipe=require('it-pipe')
72
59
73
-
pull(
74
-
pull.values(['hey, how is it going. I am the dialer']),
75
-
stream
76
-
)
60
+
// Create a duplex muxer
61
+
constmuxer=newMplex()
62
+
63
+
// Use the muxer in a pipeline
64
+
pipe(conn, muxer, conn) // conn is duplex connection to another peer
77
65
```
78
66
79
-
Now we can first run `listener.js` and then `dialer.js` to see the
80
-
following output:
67
+
`options` is an optional `Object` that may have the following properties:
68
+
69
+
*`onStream` - A function called when receiving a new stream from the remote. e.g.
70
+
```js
71
+
// Receive a new stream on the muxed connection
72
+
constonStream=stream=> {
73
+
// Read from this stream and write back to it (echo server)
74
+
pipe(
75
+
stream,
76
+
source=> (asyncfunction* () {
77
+
forawait (constdataof source) yield data
78
+
})(),
79
+
stream
80
+
)
81
+
}
82
+
constmuxer=newMplex({ onStream })
83
+
// ...
84
+
```
85
+
**Note:** The `onStream`function can be passed in place of the `options` object. i.e.
86
+
```js
87
+
new Mplex(stream=> { /* ... */ })
88
+
```
89
+
* `signal` - An [`AbortSignal`](https://developer.mozilla.org/en-US/docs/Web/API/AbortSignal) which can be used to abort the muxer, _including_ all of it's multiplexed connections. e.g.
90
+
```js
91
+
const controller = new AbortController()
92
+
const muxer = new Mplex({ signal:controller.signal })
93
+
94
+
pipe(conn, muxer, conn)
95
+
96
+
controller.abort()
97
+
```
98
+
*`maxMsgSize`- The maximum size in bytes the data field of multiplexed messages may contain (default 1MB)
99
+
100
+
### `muxer.onStream`
101
+
102
+
Use this property as an alternative to passing `onStream` as an option to the `Mplex`constructor.
103
+
104
+
### `const stream = muxer.newStream([options])`
105
+
106
+
Initiate a new stream with the remote. Returns a [duplex stream](https://gist.github.com/alanshaw/591dc7dd54e4f99338a347ef568d6ee9#duplex-it).
107
+
108
+
e.g.
81
109
82
-
*listener.js*
110
+
```js
111
+
// Create a new stream on the muxed connection
112
+
const stream = muxer.newStream()
83
113
84
-
```
85
-
$ node listener.js
86
-
[listener] listening on 9999
87
-
[listener] Got connection!
88
-
[listener] Got stream!
89
-
[listener] Received:
90
-
hey, how is it going. I am the dialer
114
+
// Use this new stream like any other duplex stream:
115
+
pipe([1, 2, 3], stream, consume)
91
116
```
92
117
93
-
*dialer.js*
118
+
In addition to `sink` and `source` properties, this stream also has the following API, that will **normally _not_ be used by stream consumers**.
94
119
95
-
```
96
-
$ node dialer.js
97
-
[dialer] opening stream
98
-
[dialer] opened stream
99
-
```
120
+
#### `stream.close()`
100
121
101
-
## Install
122
+
Closes the stream for **reading**. If iterating over the source of this stream in a `for await of` loop, it will return (exit the loop) after any buffered data has been consumed.
102
123
103
-
```sh
104
-
> npm install libp2p-mplex
105
-
```
124
+
This function is called automatically by the muxer when it receives a `CLOSE` message from the remote.
106
125
107
-
## API
126
+
The source will return normally, the sink will continue to consume.
108
127
109
-
```js
110
-
constmplex=require('libp2p-mplex')
111
-
```
128
+
#### `stream.abort([err])`
129
+
130
+
Closes the stream for **reading** _and_ **writing**. This should be called when a _local error_ has occurred.
131
+
132
+
Note, if called without an error any buffered data in the source can still be consumed and the stream will end normally.
133
+
134
+
This will cause a `RESET` message to be sent to the remote, _unless_ the sink has already ended.
135
+
136
+
The sink will return and the source will throw if an error is passed or return normally if not.
137
+
138
+
#### `stream.reset()`
139
+
140
+
Closes the stream _immediately_ for **reading** _and_ **writing**. This should be called when a _remote error_ has occurred.
141
+
142
+
This function is called automatically by the muxer when it receives a `RESET` message from the remote.
143
+
144
+
The sink will return and the source will throw.
145
+
146
+
## Contribute
147
+
148
+
The libp2p implementation in JavaScript is a work in progress. As such, there are a few things you can do right now to help out:
149
+
150
+
- Go through the modules and **check out existing issues**. This is especially useful for modules in active development. Some knowledge of IPFS/libp2p may be required, as well as the infrastructure behind it - for instance, you may need to read up on p2p and more complex operations like muxing to be able to help technically.
151
+
- **Perform code reviews**. More eyes will help a) speed the project along b) ensure quality and c) reduce possible future bugs.
0 commit comments