15185904985

Committed 22 May 2025 11:55AM UTC coverage: 87.013%. First build

Build # 15185904985

Build Type

Pull #8

github

Committed by

kevinszuchet

Commit Message

test: Run app for covergae

Pull Request Pull Request #8: feat: Community messages routing

Run Details

17 of 19 branches covered (89.47%)

Branch coverage included in aggregate %.

77 of 85 new or added lines in 4 files covered. (90.59%)

117 of 135 relevant lines covered (86.67%)

5.67 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

86.21

/src/adapters/livekit.ts

import { IBaseComponent, START_COMPONENT, STOP_COMPONENT } from '@well-known-components/interfaces'
import { AppComponents } from '../types'
import { AccessToken } from 'livekit-server-sdk'
import { DataPacketKind, RemoteParticipant, Room, RoomEvent } from '@livekit/rtc-node'
import { fromLivekitReceivedData } from '../logic/message-routing'

export type ILivekitComponent = IBaseComponent

export async function createLivekitComponent(
  components: Pick<AppComponents, 'config' | 'logs' | 'messageRouting' | 'metrics'>
): Promise<ILivekitComponent> {
  const { config, logs, messageRouting, metrics } = components
  const logger = logs.getLogger('livekit')

  const [host, apiKey, apiSecret, roomName, identityPrefix, replicaId] = await Promise.all([
    config.requireString('LIVEKIT_HOST'),
    config.requireString('LIVEKIT_API_KEY'),
    config.requireString('LIVEKIT_API_SECRET'),
    config.requireString('LIVEKIT_ROOM_NAME'),
    config.getString('LIVEKIT_IDENTITY_PREFIX'),
    config.getString('REPLICA_NUMBER')
  ])

  const identity = `${identityPrefix}-${replicaId}`

  const room = new Room()

  room.on(RoomEvent.Connected, () => {
    logger.info('Connected to LiveKit room')
    metrics.observe('livekit_connection_status', {}, 1)
  })

  room.on(RoomEvent.Reconnecting, () => {
    logger.warn('Reconnecting to LiveKit room')
    metrics.observe('livekit_connection_status', {}, 0)
  })

  room.on(RoomEvent.Reconnected, () => {
    logger.info('Reconnected to LiveKit room')
    metrics.observe('livekit_connection_status', {}, 1)
  })

  room.on(RoomEvent.Disconnected, (reason) => {
    logger.error('Disconnected from LiveKit room', { reason })
    metrics.observe('livekit_connection_status', {}, 0)
  })

  room.on(RoomEvent.DataReceived, handleDataReceived)

  async function getToken() {
    const token = new AccessToken(apiKey, apiSecret, {
      identity
    })
    token.addGrant({
      room: roomName,
      roomJoin: true,
      canPublish: true,
      canSubscribe: true,
      canPublishData: true,
      hidden: true
    })
    return token.toJwt()
  }

  async function handleDataReceived(
    payload: Uint8Array<ArrayBufferLike>,
    participant?: RemoteParticipant | undefined,
    kind?: DataPacketKind | undefined,
    topic?: string | undefined
  ) {
    if (!participant) {
      logger.error('No participant provided')
      return
    } else if (participant.identity === identity) {
      logger.warn('Received data from self')
      return
    }

    if (kind === undefined) {
      logger.error('No kind provided')
      return
    }

    if (!topic) {
      logger.error('No community id provided in the topic')
      return
    }

    logger.debug('Received data from Livekit room', {
      from: participant.identity,
      kind,
      topic
    })

    const message = fromLivekitReceivedData(payload, participant, kind, topic)
    await messageRouting.routeMessage(room, message)
  }

  return {
    [START_COMPONENT]: async () => {
      logger.debug(`Connecting identity "${identity}" to Livekit room "${roomName}"`)
      const token = await getToken()
      await room.connect(host, token)
      logger.debug('Connected to Livekit room')
    },
    [STOP_COMPONENT]: async () => {
      logger.info(`Disconnecting identity "${identity}" from Livekit room "${roomName}"`)
      await room.disconnect()
      logger.info('Disconnected from Livekit room')
    }
  }
}

1	import { IBaseComponent, START_COMPONENT, STOP_COMPONENT } from '@well-known-components/interfaces'	4✔
2	import { AppComponents } from '../types'
3	import { AccessToken } from 'livekit-server-sdk'	4✔
4	import { DataPacketKind, RemoteParticipant, Room, RoomEvent } from '@livekit/rtc-node'	4✔
5	import { fromLivekitReceivedData } from '../logic/message-routing'	4✔
6
7	export type ILivekitComponent = IBaseComponent
8
9	export async function createLivekitComponent(	4✔
10	components: Pick<AppComponents, 'config' \| 'logs' \| 'messageRouting' \| 'metrics'>
11	): Promise<ILivekitComponent> {
12	const { config, logs, messageRouting, metrics } = components	16✔
13	const logger = logs.getLogger('livekit')	16✔
14
15	const [host, apiKey, apiSecret, roomName, identityPrefix, replicaId] = await Promise.all([	16✔
16	config.requireString('LIVEKIT_HOST'),
17	config.requireString('LIVEKIT_API_KEY'),
18	config.requireString('LIVEKIT_API_SECRET'),
19	config.requireString('LIVEKIT_ROOM_NAME'),
20	config.getString('LIVEKIT_IDENTITY_PREFIX'),
21	config.getString('REPLICA_NUMBER')
22	])
23
24	const identity = `${identityPrefix}-${replicaId}`	16✔
25
26	const room = new Room()	16✔
27
28	room.on(RoomEvent.Connected, () => {	16✔
NEW 29	logger.info('Connected to LiveKit room')	×
NEW 30	metrics.observe('livekit_connection_status', {}, 1)	×
31	})
32
33	room.on(RoomEvent.Reconnecting, () => {	16✔
NEW 34	logger.warn('Reconnecting to LiveKit room')	×
NEW 35	metrics.observe('livekit_connection_status', {}, 0)	×
36	})
37
38	room.on(RoomEvent.Reconnected, () => {	16✔
NEW 39	logger.info('Reconnected to LiveKit room')	×
NEW 40	metrics.observe('livekit_connection_status', {}, 1)	×
41	})
42
43	room.on(RoomEvent.Disconnected, (reason) => {	16✔
NEW 44	logger.error('Disconnected from LiveKit room', { reason })	×
NEW 45	metrics.observe('livekit_connection_status', {}, 0)	×
46	})
47
48	room.on(RoomEvent.DataReceived, handleDataReceived)	16✔
49
50	async function getToken() {
51	const token = new AccessToken(apiKey, apiSecret, {	16✔
52	identity
53	})
54	token.addGrant({	16✔
55	room: roomName,
56	roomJoin: true,
57	canPublish: true,
58	canSubscribe: true,
59	canPublishData: true,
60	hidden: true
61	})
62	return token.toJwt()	16✔
63	}
64
65	async function handleDataReceived(
66	payload: Uint8Array<ArrayBufferLike>,
67	participant?: RemoteParticipant \| undefined,
68	kind?: DataPacketKind \| undefined,
69	topic?: string \| undefined
70	) {
71	if (!participant) {	10✔
72	logger.error('No participant provided')	2✔
73	return	2✔
74	} else if (participant.identity === identity) {	8✔
75	logger.warn('Received data from self')	2✔
76	return	2✔
77	}
78
79	if (kind === undefined) {	6✔
80	logger.error('No kind provided')	2✔
81	return	2✔
82	}
83
84	if (!topic) {	4✔
85	logger.error('No community id provided in the topic')	2✔
86	return	2✔
87	}
88
89	logger.debug('Received data from Livekit room', {	2✔
90	from: participant.identity,
91	kind,
92	topic
93	})
94
95	const message = fromLivekitReceivedData(payload, participant, kind, topic)	2✔
96	await messageRouting.routeMessage(room, message)	2✔
97	}
98
99	return {	16✔
100	[START_COMPONENT]: async () => {
101	logger.debug(`Connecting identity "${identity}" to Livekit room "${roomName}"`)	16✔
102	const token = await getToken()	16✔
103	await room.connect(host, token)	16✔
104	logger.debug('Connected to Livekit room')	16✔
105	},
106	[STOP_COMPONENT]: async () => {
107	logger.info(`Disconnecting identity "${identity}" from Livekit room "${roomName}"`)	4✔
108	await room.disconnect()	4✔
109	logger.info('Disconnected from Livekit room')	4✔
110	}
111	}
112	}

decentraland / comms-message-sfu / 15185904985

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous