Updates

test/rpc.integration.test.js

@@ -0,0 +1,99 @@
+/**
+ * Integration: HyperDHT + protomux-rpc protocol (no Docker required).
+ */
+import test from 'brittle'
+import DHT from 'hyperdht'
+import createTestnet from 'hyperdht/testnet.js'
+import ProtomuxRPC from 'protomux-rpc'
+import { PROTOCOL, Methods, Pushes } from '../shared/protocol.js'
+import { encodings } from '../shared/encodings.js'
+
+test('protomux-rpc over HyperDHT: request + push', async (t) => {
+  t.plan(5)
+
+  const testnet = await createTestnet()
+  const { bootstrap } = testnet
+
+  const serverDht = new DHT({ bootstrap })
+  const keyPair = DHT.keyPair()
+  const server = serverDht.createServer()
+
+  let pushed = false
+
+  server.on('connection', (socket) => {
+    const rpc = new ProtomuxRPC(socket, {
+      id: keyPair.publicKey,
+      protocol: PROTOCOL,
+      ...encodings,
+    })
+
+    rpc.respond(Methods.ping, encodings, () => ({ success: true, pong: Date.now() }))
+    rpc.respond(Methods.listContainers, encodings, () => ({
+      type: 'containers',
+      data: [{ Id: 'abc', Names: ['/demo'] }],
+    }))
+
+    // After client is up, push stats (simulates server broadcast)
+    setTimeout(() => {
+      rpc.event(Pushes.allStats, {
+        type: 'allStats',
+        data: [{ id: 'abc', cpu: 1.5, memory: 1024 }],
+      }, encodings)
+    }, 50)
+  })
+
+  await server.listen(keyPair)
+
+  const clientDht = new DHT({ bootstrap })
+  const socket = clientDht.connect(keyPair.publicKey)
+
+  await new Promise((resolve, reject) => {
+    const timer = setTimeout(() => reject(new Error('connect timeout')), 15000)
+    const done = () => {
+      clearTimeout(timer)
+      resolve()
+    }
+    socket.once('connect', done)
+    socket.once('open', done)
+    socket.once('error', (err) => {
+      clearTimeout(timer)
+      reject(err)
+    })
+  })
+
+  const rpc = new ProtomuxRPC(socket, {
+    id: keyPair.publicKey,
+    protocol: PROTOCOL,
+    ...encodings,
+  })
+
+  rpc.respond(Pushes.allStats, encodings, (payload) => {
+    pushed = true
+    t.is(payload.type, 'allStats')
+    t.is(payload.data[0].id, 'abc')
+    return null
+  })
+
+  const pong = await rpc.request(Methods.ping, {}, encodings)
+  t.ok(pong.success)
+
+  const list = await rpc.request(Methods.listContainers, {}, encodings)
+  t.is(list.type, 'containers')
+
+  // Wait for push
+  await new Promise((r) => setTimeout(r, 200))
+  t.ok(pushed)
+
+  socket.destroy()
+  await server.close()
+  await serverDht.destroy()
+  await clientDht.destroy()
+  await testnet.destroy()
+})
+
+test('protocol method names are stable strings', (t) => {
+  t.is(Methods.listContainers, 'listContainers')
+  t.is(Methods.startTerminal, 'startTerminal')
+  t.is(Pushes.containers, 'push:containers')
+  t.is(PROTOCOL, 'peardock/rpc')
+})