From 6eac8d97e180e2ae8046510c9c31c7aec069fbc2 Mon Sep 17 00:00:00 2001 From: talksik Date: Sat, 14 May 2022 17:03:48 -0500 Subject: [PATCH] queue working? --- packages/api/sockets/index.ts | 23 ++---- packages/core/sockets/channels.ts | 22 ++---- .../desktop/src/providers/StreamProvider.tsx | 71 +++++++------------ 3 files changed, 38 insertions(+), 78 deletions(-) diff --git a/packages/api/sockets/index.ts b/packages/api/sockets/index.ts index 7cc88b5..9b22200 100644 --- a/packages/api/sockets/index.ts +++ b/packages/api/sockets/index.ts @@ -1,9 +1,7 @@ import { ConnectToLineRequest, - RtcAnswerRequest, - RtcCallRequest, - RtcNewUserResponse, - RtcReceiveAnswerResponse, + RtcReceiveSignalResponse, + RtcSendSignalRequest, ServerRequestChannels, ServerResponseChannels, SomeoneConnectedResponse, @@ -154,21 +152,12 @@ export default function InitializeWs(io: any) { }); // tell the proper other user to create a local peer object for the one on one mesh connection - socket.on(ServerRequestChannels.RTC_CALL_REQUEST, (req: RtcCallRequest) => { - const userSocketId = userIdsToSocketIds[req.userIdToCall]; + socket.on(ServerRequestChannels.RTC_SEND_SIGNAL, (req: RtcSendSignalRequest) => { + const userSocketId = userIdsToSocketIds[req.userToCall]; io.to(userSocketId).emit( - ServerResponseChannels.RTC_NEW_USER_JOINED_RESPONSE, - new RtcNewUserResponse(userInfo.userId, req.simplePeerSignal), - ); - }); - - socket.on(ServerRequestChannels.RTC_ANSWER_REQUEST, (req: RtcAnswerRequest) => { - const userSocketId = userIdsToSocketIds[req.userIdToCall]; - - io.to(userSocketId).emit( - ServerResponseChannels.RTC_RECEIVING_ANSWER_RESPONSE, - new RtcReceiveAnswerResponse(userInfo.userId, req.simplePeerSignal), + ServerResponseChannels.RTC_RECEIVING_SIGNAL, + new RtcReceiveSignalResponse(userInfo.userId, req.simplePeerSignal), ); }); diff --git a/packages/core/sockets/channels.ts b/packages/core/sockets/channels.ts index 1d1a3e7..390bde4 100644 --- a/packages/core/sockets/channels.ts +++ b/packages/core/sockets/channels.ts @@ -42,8 +42,7 @@ export enum ServerRequestChannels { BROADCAST_TO_LINE = 'BROADCAST_TO_LINE', STOP_BROADCAST_TO_LINE = 'STOP_BROADCAST_TO_LINE', - RTC_CALL_REQUEST = 'RTC_CALL_PREFIX', - RTC_ANSWER_REQUEST = 'RTC_ANSWER_REQUEST_PREFIX', + RTC_SEND_SIGNAL = 'RTC_SEND_SIGNAL', GOING_INTO_FLOW_STATE = 'GOING_INTO_FLOW_STATE', } @@ -59,8 +58,7 @@ export enum ServerResponseChannels { SOMEONE_STOPPED_BROADCASTING = 'SOMEONE_STOPPED_BROADCASTING', // stop showing their stream tracks // sending to the correct room of tunedin folks AND also making sure it's the right event handler in the right handler for this component - RTC_NEW_USER_JOINED_RESPONSE = 'RTC_NEW_USER_JOINED_RESPONSE', - RTC_RECEIVING_ANSWER_RESPONSE = 'RTC_RECEIVING_ANSWER_RESPONSE', + RTC_RECEIVING_SIGNAL = 'RTC_RECEIVING_SIGNAL', SOMEONE_GOING_INTO_FLOW_STATE = 'SOMEONE_GOING_INTO_FLOW_STATE', } @@ -109,20 +107,12 @@ export class SocketEmitter { constructor(public channel: SocketChannels, data: T) {} } -export class RtcCallRequest { - constructor(public userIdToCall: string, public simplePeerSignal: any) {} +export class RtcSendSignalRequest { + constructor(public userToCall: string, public simplePeerSignal: any) {} } -export class RtcNewUserResponse { - constructor(public newUserId: string, public simplePeerSignal: any) {} -} - -export class RtcAnswerRequest { - constructor(public userIdToCall: string, public simplePeerSignal: any) {} -} - -export class RtcReceiveAnswerResponse { - constructor(public answererUserId: string, public simplePeerSignal: any) {} +export class RtcReceiveSignalResponse { + constructor(public senderUserId: string, public simplePeerSignal: any) {} } export class FlowStateRequest { diff --git a/packages/desktop/src/providers/StreamProvider.tsx b/packages/desktop/src/providers/StreamProvider.tsx index 874070b..3960436 100644 --- a/packages/desktop/src/providers/StreamProvider.tsx +++ b/packages/desktop/src/providers/StreamProvider.tsx @@ -26,13 +26,11 @@ import useAuth from './AuthProvider'; import { Updater, useImmer } from 'use-immer'; import useSockets from './SocketProvider'; import { - RtcAnswerRequest, - RtcCallRequest, - RtcNewUserResponse, - RtcReceiveAnswerResponse, + RtcReceiveSignalResponse, + RtcSendSignalRequest, ServerRequestChannels, + ServerResponseChannels, } from '@nirvana/core/sockets/channels'; -import { ServerResponseChannels } from '../../../core/sockets/channels'; import toast from 'react-hot-toast'; import { usePrevious } from 'react-use'; @@ -54,6 +52,7 @@ export function StreamProvider({ children }: { children: React.ReactChild }) { const { $ws } = useSockets(); const [peerMap, updatePeerMap] = useImmer({}); + const [incomingSignals, updateIncomingSignals] = useImmer<{ [peerUserId: string]: any }>({}); const [userLocalStream, setUserLocalStream] = useState(); const localStreamRef = useRef(null); @@ -103,54 +102,33 @@ export function StreamProvider({ children }: { children: React.ReactChild }) { console.log(`peer map: `, peerMap); + // on changes of the peer map ("we called someone and created local peer connection"), see if we have an incoming signal + // and set signal if we have already gotten it and remove from the signals if used useEffect(() => { - $ws.on( - ServerResponseChannels.RTC_RECEIVING_ANSWER_RESPONSE, - (res: RtcReceiveAnswerResponse) => { - console.log(`oooo some master received my call and accepted it ${JSON.stringify(res)}`); + if (Object.keys(incomingSignals).length > 0) { + Object.entries(peerMap).map(([userId, peer]) => { + if (incomingSignals[userId]) { + peer.signal(incomingSignals[userId]); - // find the peer we created earlier for this master - // ?is this okay? using the setter to get the current state? - updatePeerMap((draft) => { - const peerForAnswerer = draft[res.answererUserId]; - - if (peerForAnswerer) { - peerForAnswerer.signal(res.simplePeerSignal); - } else { - toast.error('could not find the peer we created before for this master'); - } - }); - }, - ); - - $ws.on(ServerResponseChannels.RTC_NEW_USER_JOINED_RESPONSE, (res: RtcNewUserResponse) => { - console.log('ooo newbie joined room, I guess I will accept it and send him my signal'); - - const peerForMeAndNewbie = new Peer({ - initiator: false, - trickle: false, // prevents the multiple tries on different ice servers and signal from getting called a bunch of times + updateIncomingSignals((draft) => { + delete draft[userId]; + }); + } }); + } + }, [peerMap, incomingSignals, updateIncomingSignals]); - peerForMeAndNewbie.on('signal', (signal) => { - console.log( - 'as the answerer, I am going to send back my signal so that the newbie can update his local peer for me', - ); - $ws.emit( - ServerRequestChannels.RTC_ANSWER_REQUEST, - new RtcAnswerRequest(res.newUserId, signal), - ); - }); + useEffect(() => { + $ws.on(ServerResponseChannels.RTC_RECEIVING_SIGNAL, (res: RtcReceiveSignalResponse) => { + console.log(`getting signal from someone`, res); - peerForMeAndNewbie.signal(res.simplePeerSignal); - - updatePeerMap((draft) => { - draft[res.newUserId] = peerForMeAndNewbie; + updateIncomingSignals((draft) => { + draft[res.senderUserId] = res.simplePeerSignal; }); }); return () => { - $ws.removeListener(ServerResponseChannels.RTC_RECEIVING_ANSWER_RESPONSE); - $ws.removeListener(ServerResponseChannels.RTC_NEW_USER_JOINED_RESPONSE); + $ws.removeListener(ServerResponseChannels.RTC_RECEIVING_SIGNAL); }; }, [$ws]); @@ -218,7 +196,10 @@ function StreamConnector({ localPeerConnection.on('signal', (signal) => { console.log('going to call someone'); - $ws.emit(ServerRequestChannels.RTC_CALL_REQUEST, new RtcCallRequest(peerUserId, signal)); + $ws.emit( + ServerRequestChannels.RTC_SEND_SIGNAL, + new RtcSendSignalRequest(peerUserId, signal), + ); }); // sending back the connection to the parent for everyone