#!/usr/bin/python3 import argparse import asyncio import json import logging import ssl import queue from typing import List import gi import websockets gi.require_version('Gst', '1.0') from gi.repository import Gst gi.require_version('GstWebRTC', '1.0') from gi.repository import GstWebRTC gi.require_version('GstSdp', '1.0') from gi.repository import GstSdp log = logging.getLogger(__name__) class Events: def __init__(self): self.sdp_offer = queue.Queue() self.sdp_answer = queue.Queue() self.generated_ice_candidates = queue.Queue() self.received_ice_candidates = queue.Queue() self.sdp_info = queue.Queue() # (sdp_mids, user_fragments) self.room_left = queue.Queue() class SignalingClient: def __init__(self, events: Events, uri): self.events = events self.uri = uri self.ssl_context = ssl.SSLContext() self.ssl_context.check_hostname = False self.ssl_context.verify_mode = ssl.CERT_NONE self.session_id = None async def receive(self, uri): async for msg in self.websocket: msg_json = json.loads(msg) msg_type = msg_json['Type'] msg_value = msg_json['Value'] assert self.session_id is None or self.session_id == msg_json['SessionID'] if msg_type == 'newSession': self.session_id = msg_json['SessionID'] log.info(f"New session {self.session_id}") elif msg_type == 'gotOffer': value_json = json.loads(msg_value) sdp = value_json['sdp'] log.info(f'Got SDP offer') log.debug(f'SDP offer:\n{sdp}') self.events.sdp_offer.put_nowait(sdp) elif msg_type == 'addCallerIceCandidate': value_json = json.loads(msg_value) log.info(f'Got ICE candidate') log.debug(f'ICE candidate: {value_json}') self.events.received_ice_candidates.put_nowait(value_json) elif msg_type == 'roomNotFound': log.error(f'The room was not found: {uri}') return elif msg_type == 'roomClosed': log.info(f'Oh noes, the room went away (session {self.session_id})!') self.events.room_left.put_nowait(True) return else: log.error(f'Unknown message type {msg_type}') async def send(self): sdp_mids = None user_fragments = None while True: if self.events.sdp_answer.qsize() > 0: sdp_answer = self.events.sdp_answer.get_nowait() sdp_answer_msg = json.dumps({ 'SessionID': self.session_id, 'Type': "gotAnswer", 'Value': json.dumps({ 'type': 'answer', 'sdp': sdp_answer }) }) await self.websocket.send(sdp_answer_msg) elif self.events.sdp_info.qsize() > 0: sdp_mids, user_fragments = self.events.sdp_info.get_nowait() elif self.events.generated_ice_candidates.qsize() > 0 \ and sdp_mids is not None and user_fragments is not None: mlineindex, candidate = self.events.generated_ice_candidates.get_nowait() sdp_mid = sdp_mids[mlineindex] user_fragment = user_fragments[mlineindex] icemsg_value = json.dumps({ "candidate": candidate, "sdpMid": sdp_mid, "sdpMLineIndex": mlineindex, "usernameFragment": user_fragment, }) icemsg = json.dumps({ 'SessionID': self.session_id, 'Type': 'addCalleeIceCandidate', 'Value': icemsg_value, }) log.info(f'Send ICE candidate') log.debug(f'ICE candidate: {icemsg_value}') await self.websocket.send(icemsg) else: await asyncio.sleep(0.2) async def run(self): self.session_id = None async with websockets.connect(self.uri, ssl=self.ssl_context, close_timeout=0.5) as self.websocket: receive_task = asyncio.Task(self.receive(self.uri)) send_task = asyncio.Task(self.send()) done, pending = await asyncio.wait([receive_task, send_task], return_when=asyncio.FIRST_COMPLETED) for task in pending: task.cancel() class WebRTCClient: def __init__(self, events: Events, rtmp_uri: str): self.events = events self.rtmp_uri = rtmp_uri self.webrtcbin = Gst.ElementFactory.make('webrtcbin', 'laplace') self.pipe = Gst.Pipeline.new("pipeline") Gst.Bin.do_add_element(self.pipe, self.webrtcbin) self.webrtcbin.connect('on-negotiation-needed', self.on_negotiation_needed) self.webrtcbin.connect('on-ice-candidate', self.on_ice_candidate) self.webrtcbin.connect('pad-added', self.webrtcbin_pad_added) def on_negotiation_needed(self, element): log.info('on_negotiation_needed') def on_ice_candidate(self, element, mlineindex, candidate): log.info('on_ice_candidate') self.events.generated_ice_candidates.put_nowait((mlineindex, candidate)) def webrtcbin_pad_added(self, element, pad): log.info('webrtcbin_pad_added') if pad.direction != Gst.PadDirection.SRC: return decodebin = Gst.ElementFactory.make('decodebin') decodebin.connect('pad-added', self.decodebin_pad_added) self.pipe.add(decodebin) decodebin.sync_state_with_parent() self.webrtcbin.link(decodebin) def decodebin_pad_added(self, element, pad): log.info('decodebin_pad_added') if not pad.has_current_caps(): log.info(pad, 'has no caps, ignoring') return caps = pad.get_current_caps() padsize = caps.get_size() for i in range(padsize): s = caps.get_structure(i) # Gst.Structure name = s.get_name() if name.startswith('video'): q = Gst.ElementFactory.make('queue') conv = Gst.ElementFactory.make('videoconvert') enc = Gst.ElementFactory.make('x264enc') enc.set_property('bitrate', 1000) enc.set_property('tune', 'zerolatency') capsfilter = Gst.ElementFactory.make('capsfilter') capsfilter.set_properties(Gst.Caps.from_string('video/x-h264,stream-format=(string)avc')) flvmux = Gst.ElementFactory.make('flvmux') flvmux.set_property('streamable', True) sink = Gst.ElementFactory.make('rtmpsink') sink.set_property('location', self.rtmp_uri) assert q and conv and enc and capsfilter and flvmux and sink self.pipe.add(q) self.pipe.add(conv) self.pipe.add(enc) self.pipe.add(capsfilter) self.pipe.add(flvmux) self.pipe.add(sink) q_pad_sink = q.get_static_pad('sink') assert q_pad_sink pad_link_return = pad.link(q_pad_sink) assert pad_link_return == Gst.PadLinkReturn.OK ok = q.link(conv) assert ok ok = conv.link(enc) assert ok ok = enc.link(capsfilter) assert ok ok = capsfilter.link(flvmux) assert ok ok = flvmux.link(sink) assert ok self.pipe.set_state(Gst.State.PLAYING) self.pipe.sync_children_states() elif name.startswith('audio'): q = Gst.ElementFactory.make('queue') conv = Gst.ElementFactory.make('audioconvert') resample = Gst.ElementFactory.make('audioresample') sink = Gst.ElementFactory.make('autoaudiosink') self.pipe.add(q) self.pipe.add(conv) self.pipe.add(resample) self.pipe.add(sink) self.pipe.sync_children_states() pad.link(q.get_static_pad('sink')) q.link(conv) conv.link(resample) resample.link(sink) def set_remote_desciption_done(self, gst_promise): gst_promise = Gst.Promise.new_with_change_func(self.create_answer_done) self.webrtcbin.emit('create-answer', None, gst_promise) def create_answer_done(self, gst_promise): reply = gst_promise.get_reply() answer = reply.get_value('answer') gst_promise = Gst.Promise.new_with_change_func(self.set_local_description_done) self.webrtcbin.emit('set-local-description', answer, gst_promise) sdp_message = answer.sdp mids = [sdp_message.get_media(i).get_attribute_val('mid') for i in range(sdp_message.medias_len())] user_fragments = [sdp_message.get_media(i).get_attribute_val('ice-ufrag') for i in range(sdp_message.medias_len())] sdp_answer = sdp_message.as_text() self.mids_uf = mids, user_fragments self.answer = sdp_answer def set_local_description_done(self, gst_promise): gst_promise.get_reply() sdp_answer = self.answer log.info(f'Send SDP answer') log.debug(f'SDP answer:\n{sdp_answer}') self.events.sdp_answer.put_nowait(sdp_answer) mids, user_fragments = self.mids_uf self.events.sdp_info.put_nowait((mids, user_fragments)) async def run(self): bus = Gst.Pipeline.get_bus(self.pipe) self.pipe.set_state(Gst.State.PLAYING) try: while True: if bus.have_pending(): msg = bus.pop() if msg.type == Gst.MessageType.ERROR: log.error(f'Error from gstreamer message bus: {msg.get_structure()}') return elif msg.type == Gst.MessageType.EOS: # end of stream log.info(f'Gstreamer message bus reports end of stream') return elif self.events.sdp_offer.qsize() > 0: sdp_offer = self.events.sdp_offer.get_nowait() res, sm = GstSdp.SDPMessage.new_from_text(sdp_offer) assert res == GstSdp.SDPResult.OK rd = GstWebRTC.WebRTCSessionDescription.new(GstWebRTC.WebRTCSDPType.OFFER, sm) gst_promise = Gst.Promise.new_with_change_func(self.set_remote_desciption_done) self.webrtcbin.emit('set-remote-description', rd, gst_promise) elif self.events.received_ice_candidates.qsize() > 0: ic = self.events.received_ice_candidates.get_nowait() if ic['candidate'] != '': self.webrtcbin.emit('add-ice-candidate', ic['sdpMLineIndex'], ic['candidate']) elif self.events.room_left.qsize() > 0: self.events.room_left.get_nowait() return else: await asyncio.sleep(0.1) finally: self.pipe.set_state(Gst.State.NULL) async def run_room(laplace_uri: str, rtmp_uri: str): try: events = Events() webrtc = WebRTCClient(events, rtmp_uri) signaling = SignalingClient(events, laplace_uri) webrtc_task = asyncio.Task(webrtc.run()) signaling_task = asyncio.Task(signaling.run()) done, pending = await asyncio.wait([webrtc_task, signaling_task], return_when=asyncio.FIRST_COMPLETED) for task in done: task.result() for task in pending: task.cancel() except OSError as e: print(e) async def run_room_repeated(laplace_uri: str, rtmp_uri: str, sleep_time: float): while True: 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=' default_dest = 'rtmp://localhost:1935/' default_room = 'cug' parser = argparse.ArgumentParser() parser.add_argument('-s', '--source', 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 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) rooms = args.room if len(rooms) == 0: rooms = [default_room] asyncio.run(run_rooms(args.source, args.destination, rooms, args.retry)) if __name__ == '__main__': main()