Skip to content

Commit 015513c

Browse files
lambiengcodelambiengcode
committed
fix(meeting): get client info (room, participantId) to modify their hardware status
1 parent 1a5ea7d commit 015513c

File tree

6 files changed

+113
-29
lines changed

6 files changed

+113
-29
lines changed

‎package.json

+1-1
Original file line numberDiff line numberDiff line change
@@ -39,7 +39,7 @@
3939
"socket.io": "^4.7.5",
4040
"socket.io-adapter": "^2.5.4",
4141
"socket.io-redis": "^6.1.1",
42-
"waterbus-proto": "^1.0.7",
42+
"waterbus-proto": "^1.0.9",
4343
"werift": "^0.19.1",
4444
"winston": "^3.13.0"
4545
},

‎src/domain/constants/socket_events.ts

+9-9
Original file line numberDiff line numberDiff line change
@@ -1,14 +1,14 @@
11
const SocketEvent = {
22
// Meeting
3-
publishCSS: 'BROADCAST_CSS',
4-
publishSSC: 'BROADCAST_SSC',
5-
subscribeCSS: 'REQUEST_ESTABLISH_SUBSCRIBER_CSS',
6-
answerSubscriberSSC: 'SEND_RECEIVER_SDP_SSC',
7-
answerSubscriberCSS: 'SEND_RECEIVER_SDP_CSS',
8-
publisherCandidateCSS: 'SEND_BROADCAST_CANDIDATE_CSS',
9-
publisherCandidateSSC: 'SEND_BROADCAST_CANDIDATE_SSC',
10-
subscriberCandidateCSS: 'SEND_RECEIVER_CANDIDATE_CSS',
11-
subscriberCandidateSSC: 'SEND_RECEIVER_CANDIDATE_SSC',
3+
publishCSS: 'PUBLISH_CSS',
4+
publishSSC: 'PUBLISH_SSC',
5+
subscribeCSS: 'SUBSCRIBE_CSS',
6+
answerSubscriberSSC: 'SEND_SDP_SUBSCRIBER_SSC',
7+
answerSubscriberCSS: 'SEND_SDP_SUBSCRIBER_CSS',
8+
publisherCandidateCSS: 'SEND_CANDIDATE_PUBLISHER_CSS',
9+
publisherCandidateSSC: 'SEND_CANDIDATE_PUBLISHER_SSC',
10+
subscriberCandidateCSS: 'SEND_CANDIDATE_SUBSCRIBER_CSS',
11+
subscriberCandidateSSC: 'SEND_CANDIDATE_SUBSCRIBER_SSC',
1212
leaveRoomCSS: 'LEAVE_ROOM_CSS',
1313
newParticipantSSC: 'NEW_PARTICIPANT_SSC',
1414
participantHasLeftSSC: 'PARTICIPANT_HAS_LEFT_SSC',

‎src/infrastructure/gateways/meeting/meeting.gateway.ts

+56-15
Original file line numberDiff line numberDiff line change
@@ -17,10 +17,14 @@ import { handleError } from 'src/infrastructure/helpers/error_handler';
1717
import * as webrtc from 'werift';
1818
import { SentCameraTypeDto } from './dtos/set_camera_type.dto';
1919
import { WebRTCManager } from 'src/infrastructure/services/sfu/webrtc_manager';
20+
import { MeetingGrpcService } from 'src/infrastructure/services/meeting/meeting.service';
2021

2122
@WebSocketGateway()
2223
export class MeetingGateway {
23-
constructor(private readonly rtcManager: WebRTCManager) {}
24+
constructor(
25+
private readonly rtcManager: WebRTCManager,
26+
private readonly meetingService: MeetingGrpcService,
27+
) {}
2428

2529
@WebSocketServer() private server: Server;
2630
private logger: Logger = new Logger('MeetingGateway');
@@ -33,6 +37,12 @@ export class MeetingGateway {
3337
info: { participantId: payload.participantId, roomId: payload.roomId },
3438
});
3539

40+
const participantInfo = await this.meetingService.getParticipantById({
41+
participantId: payload.participantId,
42+
});
43+
44+
if (!participantInfo) return;
45+
3646
const responsePayload = await this.rtcManager.joinRoom(
3747
client.id,
3848
payload.sdp,
@@ -43,9 +53,7 @@ export class MeetingGateway {
4353
callback: () => {
4454
client.broadcast
4555
.to(payload.roomId)
46-
.emit(SocketEvent.newParticipantSSC, {
47-
targetId: payload.participantId,
48-
});
56+
.emit(SocketEvent.newParticipantSSC, participantInfo);
4957
},
5058
},
5159
);
@@ -113,8 +121,14 @@ export class MeetingGateway {
113121

114122
@SubscribeMessage(SocketEvent.setE2eeEnabledCSS)
115123
handleSetE2eeEnable(client: ISocketClient, payload: SetHardwareStatusDto) {
116-
const roomId = client.data.roomId;
117-
const targetId = client.data.participantId;
124+
const clientInfo = this.rtcManager.getClientBySocketId({
125+
clientId: client.id,
126+
});
127+
128+
if (!clientInfo) return;
129+
130+
const roomId = clientInfo.roomId;
131+
const targetId = clientInfo.participantId;
118132

119133
if (!roomId) return;
120134

@@ -130,8 +144,14 @@ export class MeetingGateway {
130144

131145
@SubscribeMessage(SocketEvent.setCameraTypeCSS)
132146
handleSetCameraType(client: ISocketClient, payload: SentCameraTypeDto) {
133-
const roomId = client.data.roomId;
134-
const targetId = client.data.participantId;
147+
const clientInfo = this.rtcManager.getClientBySocketId({
148+
clientId: client.id,
149+
});
150+
151+
if (!clientInfo) return;
152+
153+
const roomId = clientInfo.roomId;
154+
const targetId = clientInfo.participantId;
135155

136156
if (!roomId) return;
137157

@@ -150,8 +170,14 @@ export class MeetingGateway {
150170
client: ISocketClient,
151171
payload: SetHardwareStatusDto,
152172
): any {
153-
const roomId = client.data.roomId;
154-
const targetId = client.data.participantId;
173+
const clientInfo = this.rtcManager.getClientBySocketId({
174+
clientId: client.id,
175+
});
176+
177+
if (!clientInfo) return;
178+
179+
const roomId = clientInfo.roomId;
180+
const targetId = clientInfo.participantId;
155181

156182
if (!roomId) return;
157183

@@ -170,8 +196,14 @@ export class MeetingGateway {
170196
client: ISocketClient,
171197
payload: SetHardwareStatusDto,
172198
): any {
173-
const roomId = client.data.roomId;
174-
const targetId = client.data.participantId;
199+
const clientInfo = this.rtcManager.getClientBySocketId({
200+
clientId: client.id,
201+
});
202+
203+
if (!clientInfo) return;
204+
205+
const roomId = clientInfo.roomId;
206+
const targetId = clientInfo.participantId;
175207

176208
if (!roomId) return;
177209

@@ -190,8 +222,14 @@ export class MeetingGateway {
190222
client: ISocketClient,
191223
payload: SetScreenSharingDto,
192224
): any {
193-
const roomId = client.data.roomId;
194-
const targetId = client.data.participantId;
225+
const clientInfo = this.rtcManager.getClientBySocketId({
226+
clientId: client.id,
227+
});
228+
229+
if (!clientInfo) return;
230+
231+
const roomId = clientInfo.roomId;
232+
const targetId = clientInfo.participantId;
195233

196234
if (!roomId) return;
197235

@@ -206,7 +244,7 @@ export class MeetingGateway {
206244
}
207245

208246
@SubscribeMessage(SocketEvent.leaveRoomCSS)
209-
handleLeaveRoom(client: ISocketClient, payload: any): any {
247+
async handleLeaveRoom(client: ISocketClient, payload: any) {
210248
const info = this.rtcManager.leaveRoom({ clientId: client.id });
211249

212250
if (info) {
@@ -215,6 +253,9 @@ export class MeetingGateway {
215253
});
216254

217255
client.leave(info.roomId);
256+
257+
const succeed = await this.meetingService.leaveRoom(info);
258+
this.logger.debug(`Update leave room in grpc: ${succeed}`);
218259
}
219260
}
220261
}

‎src/infrastructure/services/meeting/meeting.service.ts

+39
Original file line numberDiff line numberDiff line change
@@ -78,6 +78,45 @@ export class MeetingGrpcService implements OnModuleInit {
7878
});
7979
}
8080

81+
async getParticipantById(
82+
data: meeting.GetParticipantRequest,
83+
): Promise<meeting.GetParticipantResponse> {
84+
const dataSubject = new Subject<meeting.GetParticipantResponse>();
85+
this.$connectionSubject
86+
.pipe(
87+
switchMap((isConnected) => {
88+
if (isConnected) {
89+
return this.meetingService
90+
.getParticipantById(data)
91+
.pipe(timeout(5000));
92+
} else
93+
return throwError(() => ({
94+
code: Status.UNAVAILABLE,
95+
message: 'The service is currently unavailable',
96+
}));
97+
}),
98+
catchError((error) => {
99+
if (
100+
(error?.code === Status.UNAVAILABLE ||
101+
error?.name === 'TimeoutError') &&
102+
this.isConnected
103+
)
104+
this.$connectionSubject.next(false);
105+
return throwError(() => error);
106+
}),
107+
tap((data) => dataSubject.next(data)),
108+
tap(() => dataSubject.complete()),
109+
)
110+
.subscribe({
111+
error: (err) => dataSubject.error(err),
112+
});
113+
try {
114+
return await lastValueFrom(dataSubject.pipe(map((response) => response)));
115+
} catch (error) {
116+
this.logger.error(error.toString());
117+
}
118+
}
119+
81120
async leaveRoom(data: meeting.LeaveRoomRequest): Promise<boolean> {
82121
const dataSubject = new Subject<meeting.LeaveRoomResponse>();
83122
this.$connectionSubject

‎src/infrastructure/services/sfu/webrtc_manager.ts

+4
Original file line numberDiff line numberDiff line change
@@ -291,4 +291,8 @@ export class WebRTCManager {
291291
removeClient({ clientId }: { clientId: string }) {
292292
delete this.clients[clientId];
293293
}
294+
295+
getClientBySocketId({ clientId }): IClient | null {
296+
return this.clients[clientId];
297+
}
294298
}

‎yarn.lock

+4-4
Original file line numberDiff line numberDiff line change
@@ -6279,10 +6279,10 @@ watchpack@^2.4.0:
62796279
glob-to-regexp "^0.4.1"
62806280
graceful-fs "^4.1.2"
62816281

6282-
waterbus-proto@^1.0.7:
6283-
version "1.0.7"
6284-
resolved "https://registry.yarnpkg.com/waterbus-proto/-/waterbus-proto-1.0.7.tgz#33d46ab1a553b74cd2ee3cfa1bb5bb9d95ea5767"
6285-
integrity sha512-1e/Q+aJbn8XzZ/fzM5s604PLOqB0BS6oDXbWuwHO5bihYMQta8BQYPzGW6OSbWXhPH60GT4VLr1S/4pKG1wkkA==
6282+
waterbus-proto@^1.0.9:
6283+
version "1.0.9"
6284+
resolved "https://registry.yarnpkg.com/waterbus-proto/-/waterbus-proto-1.0.9.tgz#bd5ce407a48f4e7361cf7add680742dbf7fed4c7"
6285+
integrity sha512-6PFO921ql5EpXqOcNe7LLGKBIBhL8qufpnO8EdEv1jXzeNng3tNA9SNHrkGvEyLtLlkxZSu+QrEUFkQmCG5sCw==
62866286
dependencies:
62876287
"@grpc/grpc-js" "^1.9.3"
62886288
rxjs "^7.8.1"

0 commit comments

Comments
 (0)