import logging
import ssl
import queue
+from typing import List
import gi
import websockets
self.pipe.set_state(Gst.State.NULL)
-async def run(laplace_uri: str, rtmp_uri: str):
+async def run_room(laplace_uri: str, rtmp_uri: str):
try:
events = Events()
webrtc = WebRTCClient(events, rtmp_uri)
print(e)
-async def run_repeated(laplace_uri: str, rtmp_uri: str, sleep_time: float):
+async def run_room_repeated(laplace_uri: str, rtmp_uri: str, sleep_time: float):
while True:
- await run(laplace_uri, rtmp_uri)
+ await run_room(laplace_uri, rtmp_uri)
await asyncio.sleep(sleep_time)
+async def run_rooms(laplace_base_uri: str, rtmp_base_uri: str, rooms: List[str], retry: bool):
+ tasks = []
+ for room in rooms:
+ laplace_uri = laplace_base_uri + room # TODO: encode
+ rtmp_uri = rtmp_base_uri + room # TODO: encode
+ if retry:
+ tasks.append(run_room_repeated(laplace_uri, rtmp_uri, 2.))
+ else:
+ tasks.append(run_room(laplace_uri, rtmp_uri))
+ await asyncio.gather(*tasks)
+
+
def main():
logging.basicConfig(level=logging.DEBUG, format='%(asctime)-15s %(message)s')
- default_source = 'wss://localhost:1234/ws_connect?id=cug'
- default_dest = 'rtmp://localhost:1935/cug'
+ default_source = 'wss://localhost:1234/ws_connect?id='
+ default_dest = 'rtmp://localhost:1935/'
+ default_room = 'cug'
parser = argparse.ArgumentParser()
parser.add_argument('-s', '--source', default=default_source,
- help=f'Laplace signalling websocket URI, default: {default_source}')
+ help=f'Laplace signalling websocket base URI, default: {default_source}')
parser.add_argument('-d', '--destination', default=default_dest,
- help=f'RTMP server URI, default: {default_dest}')
+ help=f'RTMP server base URI, default: {default_dest}')
parser.add_argument('-r', '--retry', action='store_true', help=f'Retry forever if room not found or closed')
+ parser.add_argument('room', nargs='*', help=f'Room names to be used, "{default_room}" if omitted')
args = parser.parse_args()
Gst.init(None)
- if args.retry:
- job = run_repeated(args.source, args.destination, 2.)
- else:
- job = run(args.source, args.destination)
- asyncio.run(job)
+ rooms = args.room
+ if len(rooms) == 0:
+ rooms = [default_room]
+ asyncio.run(run_rooms(args.source, args.destination, rooms, args.retry))
if __name__ == '__main__':