import argparse
import asyncio
-import datetime
import json
import logging
-import pathlib
import ssl
-import sys
+import queue
from typing import Optional, List
-import websockets
-
import gi
+import websockets
gi.require_version('Gst', '1.0')
from gi.repository import Gst
gi.require_version('GstSdp', '1.0')
from gi.repository import GstSdp
+gi.require_version('GstRtspServer', '1.0')
+from gi.repository import Gst, GstRtspServer, GObject, GLib
+
log = logging.getLogger(__name__)
+class GstreamerRtspServer():
+ def __init__(self):
+ server = GstRtspServer.RTSPServer()
+ server.set_address("::")
+ server.set_service('8554') # port as string
+ factory = GstRtspServer.RTSPMediaFactory()
+ # factory.set_launch("intervideosrc ! decodebin ! theoraenc ! queue ! rtptheorapay name=pay0")
+ # factory.set_launch("intervideosrc ! decodebin ! videoconvert ! video/x-raw,format=I420 ! theoraenc ! queue ! rtptheorapay name=pay0")
+ # factory.set_launch("intervideosrc ! decodebin ! videorate ! videoconvert ! video/x-raw,format=I420,framerate=10/1 ! theoraenc ! queue ! rtptheorapay name=pay0")
+ # factory.set_launch("intervideosrc ! decodebin ! videoconvert ! video/x-raw,format=I420 ! x264enc ! queue ! rtph264pay pt=96 name=pay0")
+ factory.set_launch("intervideosrc ! decodebin ! x264enc ! queue ! rtph264pay pt=96 name=pay0")
+ # factory.set_launch("intervideosrc ! decodebin ! videorate ! videoconvert ! video/x-raw,framerate=10/1 ! x264enc ! queue ! rtph264pay pt=96 name=pay0")
+ factory.set_shared(True)
+ mountPoints = server.get_mount_points()
+ mountPoints.add_factory("/cug", factory)
+ server.attach()
+ self.server = server
+
+
class Lagarde:
def __init__(self):
self.sdp_offer: Optional[str] = None
self.websocket: Optional[websockets.client.WebSocketClientProtocol] = None
self.session_id = None
- self.received_ice_candidates = []
- self.generated_ice_candidates = []
+ self.received_ice_candidates = queue.Queue()
+ self.generated_ice_candidates = queue.Queue()
self.user_fragments: Optional[List] = None
self.mids: Optional[List] = None
self.pipe = None
def on_ice_candidate(self, element, mlineindex, candidate):
log.info('on_ice_candidate')
- self.generated_ice_candidates.append((mlineindex, candidate))
+ self.generated_ice_candidates.put_nowait((mlineindex, candidate))
def webrtcbin_pad_added(self, element, pad):
log.info('webrtcbin_pad_added')
return
caps = pad.get_current_caps()
- # assert (len(caps)) # we have a Gst.Caps object and it has no length
- # s = caps[0] # also, it's not a list
padsize = caps.get_size()
- assert(padsize > 0)
- for i in range(padsize): # pythonic?!
- s = caps.get_structure(i) # Gst.Structure
+ 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')
- sink = Gst.ElementFactory.make('autovideosink')
+ sink = Gst.ElementFactory.make('intervideosink')
self.pipe.add(q)
self.pipe.add(conv)
self.pipe.add(sink)
pad.link(q.get_static_pad('sink'))
q.link(conv)
conv.link(sink)
- 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)
+ self.pipe.set_state(Gst.State.PLAYING)
async def listen_to_gstreamer_bus(self):
- Gst.init(None)
self.webrtcbin = Gst.ElementFactory.make('webrtcbin', 'laplace')
self.pipe = Gst.Pipeline.new("pipeline")
Gst.Bin.do_add_element(self.pipe, self.webrtcbin)
try:
while True:
if bus.have_pending():
- msg = bus.pop() # Gst.Message, has to be unref'ed.
- if msg.type != Gst.MessageType.STATE_CHANGED:
- # log.info(f'Receive Gst.Message: {msg.type}, {msg.seqnum}, {msg.get_structure()}')
- # log.info(f'{webrtcbin.props.signaling_state} {webrtcbin.props.ice_gathering_state} {webrtcbin.props.ice_connection_state}')
- # Gst.Message.unref(msg)
- pass
+ 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.sdp_offer is not None:
res, sm = GstSdp.SDPMessage.new()
assert res == GstSdp.SDPResult.OK
rd = GstWebRTC.WebRTCSessionDescription.new(GstWebRTC.WebRTCSDPType.OFFER, sm)
gst_promise = Gst.Promise.new()
self.webrtcbin.emit('set-remote-description', rd, gst_promise)
- gst_promise.wait()
+ await asyncio.get_event_loop().run_in_executor(None, gst_promise.wait)
self.sdp_offer = None
- log.info('create-answer')
gst_promise = Gst.Promise.new()
self.webrtcbin.emit('create-answer', None, gst_promise)
- result = gst_promise.wait()
+ result = await asyncio.get_event_loop().run_in_executor(None, gst_promise.wait)
assert result == Gst.PromiseResult.REPLIED
reply = gst_promise.get_reply()
answer = reply.get_value('answer')
self.mids = [sdp_message.get_media(i).get_attribute_val('mid')
for i in range(sdp_message.medias_len())]
sdp_answer = sdp_message.as_text()
- log.info(sdp_answer)
+ log.info(f'Send SDP answer')
+ log.debug(f'SDP answer:\n{sdp_answer}')
sdp_answer_msg = json.dumps({
'SessionID': self.session_id,
'Type': "gotAnswer",
})
gst_promise = Gst.Promise.new()
self.webrtcbin.emit('set-local-description', answer, gst_promise)
- gst_promise.wait()
+ await asyncio.get_event_loop().run_in_executor(None, gst_promise.wait)
gst_promise.get_reply()
await self.websocket.send(sdp_answer_msg)
- elif len(self.received_ice_candidates) > 0:
- ic = self.received_ice_candidates.pop(0)
+ elif self.received_ice_candidates.qsize() > 0:
+ ic = self.received_ice_candidates.get_nowait()
if ic['candidate'] != '':
self.webrtcbin.emit('add-ice-candidate', ic['sdpMLineIndex'], ic['candidate'])
- elif len(self.generated_ice_candidates) > 0:
- mlineindex, candidate = self.generated_ice_candidates.pop(0)
+ elif self.generated_ice_candidates.qsize() > 0:
+ mlineindex, candidate = self.generated_ice_candidates.get_nowait()
+ icemsg_value = json.dumps({
+ "candidate": candidate,
+ "sdpMid": self.mids[mlineindex],
+ "sdpMLineIndex": mlineindex,
+ "usernameFragment": self.user_fragments[mlineindex],
+ })
icemsg = json.dumps({
'SessionID': self.session_id,
'Type': 'addCalleeIceCandidate',
- 'Value': json.dumps({
- "candidate": candidate,
- "sdpMid": self.mids[mlineindex],
- "sdpMLineIndex": mlineindex,
- "usernameFragment": self.user_fragments[mlineindex],
- })
+ 'Value': icemsg_value,
})
- log.info(f'send_ice_candidate_message with {icemsg}')
+ log.info(f'Send ICE candidate')
+ log.debug(f'ICE candidate: {icemsg_value}')
await self.websocket.send(icemsg)
else:
finally:
self.pipe.set_state(Gst.State.NULL)
- async def talk_to_websocket(self, uri):
- ssl_context = ssl.SSLContext()
- ssl_context.check_hostname = False
- ssl_context.verify_mode = ssl.CERT_NONE
- async with websockets.connect(uri, ssl=ssl_context) as self.websocket:
+ async def talk_to_websocket(self, uri, ssl_context):
+ async with websockets.connect(uri, ssl=ssl_context, close_timeout=0.5) as self.websocket:
async for msg in self.websocket:
msg_json = json.loads(msg)
msg_type = msg_json['Type']
msg_value = msg_json['Value']
- self.session_id = msg_json['SessionID']
- log.info(f"receive for session {self.session_id} type {msg_type}")
+ assert self.session_id is None or self.session_id == msg_json['SessionID']
if msg_type == 'newSession':
- pass
+ 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'SDP: {sdp}')
+ log.info(f'Got SDP offer')
+ log.debug(f'SDP offer:\n{sdp}')
self.sdp_offer = sdp
elif msg_type == 'addCallerIceCandidate':
value_json = json.loads(msg_value)
- log.info(f'ICE: {value_json}')
- self.received_ice_candidates.append(value_json)
+ log.info(f'Got ICE candidate')
+ log.debug(f'ICE candidate: {value_json}')
+ self.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('Oh noes, the room went away!')
- # and here we should clean up
+ log.info(f'Oh noes, the room went away (session {self.session_id})!')
+ self.session_id = None
+ return
else:
log.error(f'Unknown message type {msg_type}')
+ async def talk_to_signaling_server(self, uri):
+ ssl_context = ssl.SSLContext()
+ ssl_context.check_hostname = False
+ ssl_context.verify_mode = ssl.CERT_NONE
+ while True:
+ await self.talk_to_websocket(uri, ssl_context)
+ await asyncio.sleep(0.1)
+
async def run(self, uri):
- talk_to_websocket_task = asyncio.Task(self.talk_to_websocket(uri))
- listen_to_gstreamer_bus_task = asyncio.Task(self.listen_to_gstreamer_bus())
- done, pending = await asyncio.wait(
- [talk_to_websocket_task, listen_to_gstreamer_bus_task],
- return_when=asyncio.FIRST_COMPLETED)
- for d in done:
- d.result()
- for p in pending:
- p.cancel()
+ try:
+ talk_to_signaling_server_task = asyncio.Task(self.talk_to_signaling_server(uri))
+ listen_to_gstreamer_bus_task = asyncio.Task(self.listen_to_gstreamer_bus())
+ main_loop = asyncio.Task(gstreamer_main_loop())
+ done, pending = await asyncio.wait(
+ [talk_to_signaling_server_task, listen_to_gstreamer_bus_task, main_loop],
+ return_when=asyncio.FIRST_COMPLETED)
+ for d in done:
+ d.result()
+ for p in pending:
+ p.cancel()
+ except OSError as e:
+ print(e)
+
+
+async def gstreamer_main_loop():
+ """Does the equivalent of the following lines in an async friendly way:
+ loop = GLib.MainLoop()
+ loop.run()
+ """
+ gst_loop = GLib.MainLoop()
+ context = gst_loop.get_context()
+ while True:
+ events_dispatched = context.iteration(False)
+ await asyncio.sleep(0. if events_dispatched else 0.01)
def main():
parser.add_argument('--uri', default='wss://localhost:1234/ws_connect?id=cug',
help='Signalling server URI')
args = parser.parse_args()
+
+ Gst.init(None)
+ rtsp = GstreamerRtspServer()
lagarde = Lagarde()
asyncio.run(lagarde.run(args.uri), debug=True)
ToastFreeware / toast / stream2beamer.git / blobdiff