stream format encryption error - unable to setup 2 nodes communication

[fgtest1]

HI,
I am trying to solve this communication between 2 nodes but it fails at the stream format or something else.
here is the error messages and code
C:\APPS3>node start-nodes-Test1.js
Node 1 started with Peer ID: 12D3KooWSpGLzdws6AfmBfAfyx9qk5o56XJUZ6w2Rq6kGMM7HNop
Node 2 started with Peer ID: 12D3KooWC3ZgGRDhsYcHZcq5NvYxpJ2PGuYxrHFQBRKJGKdpmFWW
Node 1 Listening Addresses: [
'/ip4/127.0.0.1/tcp/3001/p2p/12D3KooWSpGLzdws6AfmBfAfyx9qk5o56XJUZ6w2Rq6kGMM7HNop'
]
Node 2 Listening Addresses: [
'/ip4/127.0.0.1/tcp/3002/p2p/12D3KooWC3ZgGRDhsYcHZcq5NvYxpJ2PGuYxrHFQBRKJGKdpmFWW'
]
Both nodes have registered the /chat/1.0.0 protocol.
Node 2 discovered peer: 12D3KooWSpGLzdws6AfmBfAfyx9qk5o56XJUZ6w2Rq6kGMM7HNop
Node 1 discovered peer: 12D3KooWC3ZgGRDhsYcHZcq5NvYxpJ2PGuYxrHFQBRKJGKdpmFWW
Attempting to dial protocol /chat/1.0.0...
Node 1 connected to peer: 12D3KooWC3ZgGRDhsYcHZcq5NvYxpJ2PGuYxrHFQBRKJGKdpmFWW
Node 2 connected to peer: 12D3KooWSpGLzdws6AfmBfAfyx9qk5o56XJUZ6w2Rq6kGMM7HNop
Node 2 received a message
Node 2 received an invalid stream object: undefined
Dial result: MplexStream {}
Stream object: MplexStream {}
Stream methods: {
write: 'undefined',
read: 'undefined',
sink: 'undefined',
source: 'undefined'
}
Invalid stream object: MplexStream {}

*** the code
import { createNode } from './create-node.js'
import { pipe } from 'it-pipe'
import { toString as uint8ArrayToString } from 'uint8arrays/to-string'

async function main() {
try {
// Create two nodes
const node1 = await createNode(3001)
const node2 = await createNode(3002)

// Start the nodes
await node1.start()
await node2.start()

// Log Peer IDs and listening addresses
console.log('Node 1 started with Peer ID:', node1.peerId.toString())
console.log('Node 2 started with Peer ID:', node2.peerId.toString())

console.log('Node 1 Listening Addresses:', node1.getMultiaddrs().map(addr => addr.toString()))
console.log('Node 2 Listening Addresses:', node2.getMultiaddrs().map(addr => addr.toString()))

// Register the /chat/1.0.0 protocol on both nodes
const handleProtocol = (node, nodeName) => {
node.handle('/chat/1.0.0', async ({ stream }) => {
console.log(${nodeName} received a message)

// Validate the stream object
if (!stream || typeof stream.source !== 'object') {
  console.error(`${nodeName} received an invalid stream object:`, stream)
  return
}

console.log(`${nodeName} stream object is valid:`, stream)

// Read raw data from the stream
try {
  for await (const chunk of stream.source) {
    console.log(`${nodeName} received chunk:`, chunk)
    const message = uint8ArrayToString(chunk)
    console.log(`> ${message}`)
  }
} catch (error) {
  console.error(`${nodeName} error reading stream:`, error)
}

})
}

handleProtocol(node1, 'Node 1')
handleProtocol(node2, 'Node 2')

console.log('Both nodes have registered the /chat/1.0.0 protocol.')

// Set up event listeners for peer discovery and connection
node1.addEventListener('peer:discovery', async (event) => {
const peerId = event.detail.peerId || event.detail.id
if (!peerId) {
console.error('Discovered peer has no PeerId:', event.detail)
return
}

console.log(Node 1 discovered peer: ${peerId.toString()})

setTimeout(async () => {
try {
console.log('Attempting to dial protocol /chat/1.0.0...')
const dialResult = await node1.dialProtocol(peerId, '/chat/1.0.0')
console.log('Dial result:', dialResult)

  // Validate the stream object
  const stream = dialResult.stream || dialResult // Check if stream is nested
  console.log('Stream object:', stream)
  console.log('Stream methods:', {
    write: typeof stream?.write,
    read: typeof stream?.read,
    sink: typeof stream?.sink,
    source: typeof stream?.source
  })

  if (!stream || typeof stream.write !== 'function' || typeof stream.read !== 'function') {
    console.error('Invalid stream object:', stream)
    return
  }

  console.log('Stream object is valid:', stream)

  // Send stdin to the stream and read responses
  process.stdin.on('data', (buf) => {
    try {
      stream.write(buf)
    } catch (error) {
      console.error('Error writing to stream:', error)
    }
  })

  process.stdin.on('end', () => {
    console.log('Closing stdin input.')
    stream.end()
  })

  process.stdin.on('error', (err) => {
    console.error('Error reading from stdin:', err)
  })
} catch (error) {
  console.error('Error during dialProtocol:', error)
}

}, 5000) // Increase delay to 5 seconds
})

node2.addEventListener('peer:discovery', (event) => {
const peerId = event.detail.peerId || event.detail.id
console.log(Node 2 discovered peer: ${peerId?.toString()})
})

node1.addEventListener('peer:connect', (event) => {
console.log('Node 1 connected to peer:', event.detail.toString())
})

node2.addEventListener('peer:connect', (event) => {
console.log('Node 2 connected to peer:', event.detail.toString())
})
} catch (error) {
console.error('Error starting nodes:', error)
}
}

main()

*** the create-node.js

import { createLibp2p } from 'libp2p'
import { tcp } from '@libp2p/tcp'
import { mplex } from '@libp2p/mplex'
import { noise } from '@chainsafe/libp2p-noise'
import { mdns } from '@libp2p/mdns'

export async function createNode(port) {
const node = await createLibp2p({
addresses: {
listen: [/ip4/127.0.0.1/tcp/${port}]
},
transports: [tcp()],
streamMuxers: [mplex()],
connectionEncrypters: [noise()], // Renamed from connectionEncryption
peerDiscovery: [mdns()] // Enable peer discovery
})

return node
}

*** libs
C:\APPS3>npm list
[email protected] C:\APPS3
+-- @chainsafe/[email protected]
+-- @libp2p/[email protected]
+-- @libp2p/[email protected]
+-- @libp2p/[email protected]
+-- @libp2p/[email protected]
+-- @libp2p/[email protected]
+-- @libp2p/[email protected]
+-- @libp2p/[email protected]
+-- @libp2p/[email protected]
+-- @libp2p/[email protected]
+-- @libp2p/[email protected]
+-- @multiformats/[email protected]
+-- [email protected]
+-- [email protected]
+-- [email protected]
+-- [email protected]
`-- [email protected]

[tabcat]

node.handle('/chat/1.0.0', async ({ stream }) => {

The handler destructures the stream parameter. So { stream } here is actually Stream.stream which is undefined.
https://libp2p.github.io/js-libp2p/interfaces/_libp2p_interface.StreamHandler.html

I also don't see write or read properties on the Stream interface. It's possible this was changed recently, I'm not sure.
https://libp2p.github.io/js-libp2p/interfaces/_libp2p_interface.Stream.html

I would recommend referring to and forking the example repo https://github.com/libp2p/js-libp2p-example-custom-protocols
Also typescript would have helped a lot here. Worth the trouble!