Skip to content

Commit

Permalink
fix: make mss check logger before use (#2261) (#2274)
Browse files Browse the repository at this point in the history 
Check logger is present before use. Fixes a regression introduced during the 1.0 release where mss didn't go out as a major - reintroduces backwards compatibility.
  • Loading branch information
@achingbrain
achingbrain authored Dec 1, 2023
1 parent bca8d6e commit cf96369
Show file tree
Hide file tree
Showing 3 changed files with 33 additions and 33 deletions.
22 changes: 11 additions & 11 deletions packages/multistream-select/src/handle.ts
View file
Original file line number Diff line number Diff line change
Expand Up @@ -55,7 +55,7 @@ import type { Duplex } from 'it-stream-types'
*/
export async function handle <Stream extends Duplex<any, any, any>> (stream: Stream, protocols: string | string[], options: MultistreamSelectInit): Promise<ProtocolStream<Stream>> {
protocols = Array.isArray(protocols) ? protocols : [protocols]
options.log.trace('handle: available protocols %s', protocols)
options?.log?.trace('handle: available protocols %s', protocols)

const lp = lpStream(stream, {
...options,
Expand All @@ -64,21 +64,21 @@ export async function handle <Stream extends Duplex<any, any, any>> (stream: Str
})

while (true) {
options?.log.trace('handle: reading incoming string')
options?.log?.trace('handle: reading incoming string')
const protocol = await multistream.readString(lp, options)
options.log.trace('handle: read "%s"', protocol)
options?.log?.trace('handle: read "%s"', protocol)

if (protocol === PROTOCOL_ID) {
options.log.trace('handle: respond with "%s" for "%s"', PROTOCOL_ID, protocol)
options?.log?.trace('handle: respond with "%s" for "%s"', PROTOCOL_ID, protocol)
await multistream.write(lp, uint8ArrayFromString(`${PROTOCOL_ID}\n`), options)
options.log.trace('handle: responded with "%s" for "%s"', PROTOCOL_ID, protocol)
options?.log?.trace('handle: responded with "%s" for "%s"', PROTOCOL_ID, protocol)
continue
}

if (protocols.includes(protocol)) {
options.log.trace('handle: respond with "%s" for "%s"', protocol, protocol)
options?.log?.trace('handle: respond with "%s" for "%s"', protocol, protocol)
await multistream.write(lp, uint8ArrayFromString(`${protocol}\n`), options)
options.log.trace('handle: responded with "%s" for "%s"', protocol, protocol)
options?.log?.trace('handle: responded with "%s" for "%s"', protocol, protocol)

return { stream: lp.unwrap(), protocol }
}
Expand All @@ -90,14 +90,14 @@ export async function handle <Stream extends Duplex<any, any, any>> (stream: Str
uint8ArrayFromString('\n')
)

options.log.trace('handle: respond with "%s" for %s', protocols, protocol)
options?.log?.trace('handle: respond with "%s" for %s', protocols, protocol)
await multistream.write(lp, protos, options)
options.log.trace('handle: responded with "%s" for %s', protocols, protocol)
options?.log?.trace('handle: responded with "%s" for %s', protocols, protocol)
continue
}

options.log('handle: respond with "na" for "%s"', protocol)
options?.log?.('handle: respond with "na" for "%s"', protocol)
await multistream.write(lp, uint8ArrayFromString('na\n'), options)
options.log('handle: responded with "na" for "%s"', protocol)
options?.log?.('handle: responded with "na" for "%s"', protocol)
}
}
2 changes: 1 addition & 1 deletion packages/multistream-select/src/multistream.ts
View file
Original file line number Diff line number Diff line change
Expand Up @@ -30,7 +30,7 @@ export async function read (reader: LengthPrefixedStream<Duplex<AsyncGenerator<U
const buf = await reader.read(options)

if (buf.byteLength === 0 || buf.get(buf.byteLength - 1) !== NewLine[0]) {
options?.log.error('Invalid mss message - missing newline', buf)
options?.log?.error('Invalid mss message - missing newline', buf)
throw new CodeError('missing newline', 'ERR_INVALID_MULTISTREAM_SELECT_MESSAGE')
}

Expand Down
42 changes: 21 additions & 21 deletions packages/multistream-select/src/select.ts
View file
Original file line number Diff line number Diff line change
Expand Up @@ -78,20 +78,20 @@ export async function select <Stream extends SelectStream> (stream: Stream, prot
throw new Error('At least one protocol must be specified')
}

options?.log.trace('select: write ["%s", "%s"]', PROTOCOL_ID, protocol)
options?.log?.trace('select: write ["%s", "%s"]', PROTOCOL_ID, protocol)
const p1 = uint8ArrayFromString(`${PROTOCOL_ID}\n`)
const p2 = uint8ArrayFromString(`${protocol}\n`)
await multistream.writeAll(lp, [p1, p2], options)

options?.log.trace('select: reading multistream-select header')
options?.log?.trace('select: reading multistream-select header')
let response = await multistream.readString(lp, options)
options?.log.trace('select: read "%s"', response)
options?.log?.trace('select: read "%s"', response)

// Read the protocol response if we got the protocolId in return
if (response === PROTOCOL_ID) {
options?.log.trace('select: reading protocol response')
options?.log?.trace('select: reading protocol response')
response = await multistream.readString(lp, options)
options?.log.trace('select: read "%s"', response)
options?.log?.trace('select: read "%s"', response)
}

// We're done
Expand All @@ -101,11 +101,11 @@ export async function select <Stream extends SelectStream> (stream: Stream, prot

// We haven't gotten a valid ack, try the other protocols
for (const protocol of protocols) {
options?.log.trace('select: write "%s"', protocol)
options?.log?.trace('select: write "%s"', protocol)
await multistream.write(lp, uint8ArrayFromString(`${protocol}\n`), options)
options?.log.trace('select: reading protocol response')
options?.log?.trace('select: reading protocol response')
const response = await multistream.readString(lp, options)
options?.log.trace('select: read "%s" for "%s"', response, protocol)
options?.log?.trace('select: read "%s" for "%s"', response, protocol)

if (response === protocol) {
return { stream: lp.unwrap(), protocol }
Expand Down Expand Up @@ -163,7 +163,7 @@ function optimisticSelect <Stream extends SelectStream> (stream: Stream, protoco
if (!sentProtocol) {
sendingProtocol = true

options?.log.trace('optimistic: write ["%s", "%s", data(%d)] in sink', PROTOCOL_ID, protocol, buf.byteLength)
options?.log?.trace('optimistic: write ["%s", "%s", data(%d)] in sink', PROTOCOL_ID, protocol, buf.byteLength)

const protocolString = `${protocol}\n`

Expand All @@ -176,7 +176,7 @@ function optimisticSelect <Stream extends SelectStream> (stream: Stream, protoco
buf
).subarray()

options?.log.trace('optimistic: wrote ["%s", "%s", data(%d)] in sink', PROTOCOL_ID, protocol, buf.byteLength)
options?.log?.trace('optimistic: wrote ["%s", "%s", data(%d)] in sink', PROTOCOL_ID, protocol, buf.byteLength)

sentProtocol = true
sendingProtocol = false
Expand All @@ -198,7 +198,7 @@ function optimisticSelect <Stream extends SelectStream> (stream: Stream, protoco

async function negotiate (): Promise<void> {
if (negotiating) {
options?.log.trace('optimistic: already negotiating %s stream', protocol)
options?.log?.trace('optimistic: already negotiating %s stream', protocol)
await doneNegotiating.promise
return
}
Expand All @@ -208,13 +208,13 @@ function optimisticSelect <Stream extends SelectStream> (stream: Stream, protoco
try {
// we haven't sent the protocol yet, send it now
if (!sentProtocol) {
options?.log.trace('optimistic: doing send protocol for %s stream', protocol)
options?.log?.trace('optimistic: doing send protocol for %s stream', protocol)
await doSendProtocol()
}

// if we haven't read the protocol response yet, do it now
if (!readProtocol) {
options?.log.trace('optimistic: doing read protocol for %s stream', protocol)
options?.log?.trace('optimistic: doing read protocol for %s stream', protocol)
await doReadProtocol()
}
} finally {
Expand All @@ -233,12 +233,12 @@ function optimisticSelect <Stream extends SelectStream> (stream: Stream, protoco
sendingProtocol = true

try {
options?.log.trace('optimistic: write ["%s", "%s", data] in source', PROTOCOL_ID, protocol)
options?.log?.trace('optimistic: write ["%s", "%s", data] in source', PROTOCOL_ID, protocol)
await lp.writeV([
uint8ArrayFromString(`${PROTOCOL_ID}\n`),
uint8ArrayFromString(`${protocol}\n`)
])
options?.log.trace('optimistic: wrote ["%s", "%s", data] in source', PROTOCOL_ID, protocol)
options?.log?.trace('optimistic: wrote ["%s", "%s", data] in source', PROTOCOL_ID, protocol)
} finally {
sentProtocol = true
sendingProtocol = false
Expand All @@ -255,15 +255,15 @@ function optimisticSelect <Stream extends SelectStream> (stream: Stream, protoco
readingProtocol = true

try {
options?.log.trace('optimistic: reading multistream select header')
options?.log?.trace('optimistic: reading multistream select header')
let response = await multistream.readString(lp, options)
options?.log.trace('optimistic: read multistream select header "%s"', response)
options?.log?.trace('optimistic: read multistream select header "%s"', response)

if (response === PROTOCOL_ID) {
response = await multistream.readString(lp, options)
}

options?.log.trace('optimistic: read protocol "%s", expecting "%s"', response, protocol)
options?.log?.trace('optimistic: read protocol "%s", expecting "%s"', response, protocol)

if (response !== protocol) {
throw new CodeError('protocol selection failed', 'ERR_UNSUPPORTED_PROTOCOL')
Expand All @@ -279,7 +279,7 @@ function optimisticSelect <Stream extends SelectStream> (stream: Stream, protoco
// make sure we've done protocol negotiation before we read stream data
await negotiate()

options?.log.trace('optimistic: reading data from "%s" stream', protocol)
options?.log?.trace('optimistic: reading data from "%s" stream', protocol)
yield * lp.unwrap().source
})()

Expand All @@ -291,7 +291,7 @@ function optimisticSelect <Stream extends SelectStream> (stream: Stream, protoco
// this before closing the readable end of the stream
if (!negotiated) {
await negotiate().catch(err => {
options?.log.error('could not negotiate protocol before close read', err)
options?.log?.error('could not negotiate protocol before close read', err)
})
}

Expand All @@ -308,7 +308,7 @@ function optimisticSelect <Stream extends SelectStream> (stream: Stream, protoco
// this before closing the writable end of the stream
if (!negotiated) {
await negotiate().catch(err => {
options?.log.error('could not negotiate protocol before close write', err)
options?.log?.error('could not negotiate protocol before close write', err)
})
}

Expand Down

0 comments on commit cf96369

Please sign in to comment.