feat: add get subscribers for pubsub topics

packages/libp2p-daemon-client/src/index.ts

@@ -295,6 +295,7 @@ export interface PubSubClient {
   publish: (topic: string, data: Uint8Array) => Promise<void>
   subscribe: (topic: string) => AsyncIterable<PSMessage>
   getTopics: () => Promise<string[]>
+  getSubscribers: (topic: string) => Promise<PeerId[]>
 }
 
 export interface DaemonClient {

packages/libp2p-daemon-client/src/pubsub.ts

@@ -6,6 +6,8 @@ import {
   PSMessage
 } from '@libp2p/daemon-protocol'
 import type { DaemonClient } from './index.js'
+import type { PeerId } from '@libp2p/interface-peer-id'
+import { peerIdFromBytes } from '@libp2p/peer-id'
 
 export class Pubsub {
   private readonly client: DaemonClient
@@ -123,4 +125,38 @@ export class Pubsub {
       yield PSMessage.decode(message)
     }
   }
+
+  async getSubscribers (topic: string): Promise<PeerId[]> {
+    if (typeof topic !== 'string') {
+      throw new CodeError('invalid topic received', 'ERR_INVALID_TOPIC')
+    }
+
+    const sh = await this.client.send({
+      type: Request.Type.PUBSUB,
+      pubsub: {
+        type: PSRequest.Type.LIST_PEERS,
+        topic
+      }
+    })
+
+    const message = await sh.read()
+
+    if (message == null) {
+      throw new CodeError('Empty response from remote', 'ERR_EMPTY_RESPONSE')
+    }
+
+    const response = Response.decode(message)
+
+    await sh.close()
+
+    if (response.type !== Response.Type.OK) {
+      throw new CodeError(response.error?.msg ?? 'Pubsub get subscribers failed', 'ERR_PUBSUB_GET_SUBSCRIBERS_FAILED')
+    }
+
+    if (response.pubsub == null || response.pubsub.topics == null) {
+      throw new CodeError('Invalid response', 'ERR_PUBSUB_GET_SUBSCRIBERS_FAILED')
+    }
+
+    return response.pubsub.peerIDs.map(buf => peerIdFromBytes(buf))
+  }
 }

packages/libp2p-daemon-client/test/pubsub.spec.ts

@@ -8,6 +8,7 @@ import { createClient, DaemonClient } from '../src/index.js'
 import { multiaddr } from '@multiformats/multiaddr'
 import { StubbedInstance, stubInterface } from 'sinon-ts'
 import type { PubSub } from '@libp2p/interface-pubsub'
+import { peerIdFromString } from '@libp2p/peer-id'
 
 const defaultMultiaddr = multiaddr('/ip4/0.0.0.0/tcp/12345')
 
@@ -88,4 +89,33 @@ describe('daemon pubsub client', function () {
       await expect(client.pubsub.publish(topic, data)).to.eventually.be.rejectedWith(/Urk!/)
     })
   })
+
+  describe('getSubscribers', () => {
+    it('should get empty list of topics when no subscriptions exist', async () => {
+      pubsub.getSubscribers.returns([])
+
+      const topic = 'test-topic'
+      const topics = await client.pubsub.getSubscribers(topic)
+
+      expect(topics).to.have.lengthOf(0)
+    })
+
+    it('should get a list with a peer when subscribed', async () => {
+      const topic = 'test-topic'
+      const peer = peerIdFromString('12D3KooWKnQbfH5t1XxJW5FBoMGNjmC9LTSbDdRJxtYj2bJV5XfP')
+      pubsub.getSubscribers.withArgs(topic).returns([peer])
+
+      const peers = await client.pubsub.getSubscribers(topic)
+
+      expect(peers).to.have.lengthOf(1)
+      expect(peers[0].toString()).to.equal(peer.toString())
+    })
+
+    it('should error if receive an error message', async () => {
+      const topic = 'test-topic'
+      pubsub.getSubscribers.throws(new Error('Urk!'))
+
+      await expect(client.pubsub.getSubscribers(topic)).to.eventually.be.rejectedWith(/Urk!/)
+    })
+  })
 })

packages/libp2p-daemon-server/src/index.ts

@@ -297,6 +297,13 @@ export class Server implements Libp2pServer {
 
           yield * this.pubsubOperations.publish(request.topic, request.data)
           return
+        case PSRequest.Type.LIST_PEERS:
+          if (request.topic == null) {
+            throw new Error('Invalid request')
+          }
+
+          yield * this.pubsubOperations.listPeers(request.topic)
+          return
         default:
           throw new Error('ERR_INVALID_REQUEST_TYPE')
       }

packages/libp2p-daemon-server/src/pubsub.ts

@@ -84,4 +84,18 @@ export class PubSubOperations {
       yield ErrorResponse(err)
     }
   }
+
+  async * listPeers (topic: string): AsyncGenerator<Uint8Array, void, undefined> {
+    try {
+      yield OkResponse({
+        pubsub: {
+          topics: [topic],
+          peerIDs: this.pubsub.getSubscribers(topic).map(peer => peer.toBytes())
+        }
+      })
+    } catch (err: any) {
+      log.error(err)
+      yield ErrorResponse(err)
+    }
+  }
 }