Invalid stream object: MplexStream {}

[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]

[MarcoPolo]

looks like JS?

[fgtest1]

can you explain please js?
do you mean the version or something

[fgtest1]

I asked grok..
SolutionGiven the consistent undefined stream issue with both mplex and yamux (from previous runs), the most likely culprit is a bug or API change in [email protected]. To resolve this, we’ll:Downgrade to [email protected]: This version is more stable and widely used, with a well-documented dialProtocol API that returns { stream, protocol } with a valid duplex stream (having sink and source).
Update Dependencies: Install compatible versions of related packages to ensure stability.
Retain Debugging: Keep the detailed logging to verify stream creation and protocol negotiation.

will test and advise here - the answer of 'looks like JS? is not clear or provide any guidance.

[MarcoPolo]

Looks like you're using Javascript, but you're commenting on a the Go implementation of libp2p. Hence my confusion. You might have better luck at https://github.com/libp2p/js-libp2p

[fgtest1]

ok got it -- wrong space thanks