client.js

const fetchResultViaPeer = async (peer, query) => {
  if (peer.id == libp2p.peerId) return // Don't send the query to ourself. This shouldn't happen in the first place, right?
  if (!peer.protocols.includes(ConstellationProtocols.PROTOCOL)) return // Only contact peers that are in the Wham-Bam constellation

  const connection = window.libp2p.connectionManager.get(peer.id)
  if (!connection) return

  console.log(`[FETCH VIA PEER] PIPE Sending Query to peer ID: ${peer.id}`)
  const { stream } = await connection.newStream([ConstellationProtocols.PROTOCOL])


  const result = await pipe(
    // Source data
    [query],
    // Write to the stream, and pass its output to the next function
    stream,
    // Sink function
    async function collect (source) {
      const vals = []
      // For each chunk of data
      for await (const val of source) {
        vals.push(val)
        // Output the data
        console.log('[Fetch Via Peer PIPE] received stream response:', vals.toString())
        return vals
        // return data.toString();
      }
      return vals
    }
  )

  const hackResult = await result.toString().split('///WTF-EOF')[0]
  console.log('[FETCHED VIA PEER] Result: ', hackResult)
  return hackResult
};


// Send request to all connected peers and take the first responder.
let peerRequestsPromises = [];
window.libp2p.peerStore.peers.forEach(peer =>
  peerRequestsPromises.push(fetchResultViaPeer(peer, payload))
)

const resBody = await Promise.any(peerRequestsPromises);

protocol.js

async function fetcher(query)  {
  console.log(`[fetcher()] Started query: https://www.localhost.com?${query}&format=json`)
  const resp = await fetch(`https://www.localhost.com?${query}&format=json`)
  const text = await resp.text();
  console.log(`[fetcher()] Received text: ${text}`)
  return text + "///WTF-EOF -------------------------------------------------------" // HACK because I can't figure out why libp2p stream is truncating
}


async function handler ({ connection, stream }) {
  await pipe(
    stream.source,
    async function collect (source) {
      const vals = []
      for await (const val of source) {
        vals.push(val)
      }
      return vals
    },
    async function executeSearch (source) {
      const query = await source
      const result = []
      const fetchResult = await fetcher(query.toString())
      result.push(fetchResult)
      console.log(`My Reply back is: ${result}`)
      const retRes = [[result],[]] //hack attempt didn't work, result still truncated
      return result
    },
    stream.sink
  );
}