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 ! 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)
self.webrtcbin = None
def on_negotiation_needed(self, element):
- log.debug('on_negotiation_needed')
+ log.info('on_negotiation_needed')
def on_ice_candidate(self, element, mlineindex, candidate):
- log.debug('on_ice_candidate')
+ log.info('on_ice_candidate')
self.generated_ice_candidates.put_nowait((mlineindex, candidate))
def webrtcbin_pad_added(self, element, pad):
- log.debug('webrtcbin_pad_added')
+ log.info('webrtcbin_pad_added')
if pad.direction != Gst.PadDirection.SRC:
return
decodebin = Gst.ElementFactory.make('decodebin')
self.webrtcbin.link(decodebin)
def decodebin_pad_added(self, element, pad):
- log.debug('decodebin_pad_added')
+ log.info('decodebin_pad_added')
if not pad.has_current_caps():
- log.debug(pad, 'has no caps, ignoring')
+ 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
+ s = caps.get_structure(i) # Gst.Structure
name = s.get_name()
if name.startswith('video'):
q = Gst.ElementFactory.make('queue')
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):
self.webrtcbin = Gst.ElementFactory.make('webrtcbin', 'laplace')
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(f'Send SDP answer:\n{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",
'Type': 'addCalleeIceCandidate',
'Value': icemsg_value,
})
- log.info(f'Send ICE candidate: {icemsg_value}')
+ 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):
- 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:\n{sdp}')
- self.sdp_offer = sdp
- elif msg_type == 'addCallerIceCandidate':
- value_json = json.loads(msg_value)
- log.info(f'Got 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(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_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']
+ 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.sdp_offer = 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.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.session_id = None
+ return
+ else:
+ log.error(f'Unknown message type {msg_type}')
- async def run(self, uri):
+ 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):
try:
- async with websockets.connect(uri, ssl=ssl_context) as self.websocket:
- talk_to_websocket_task = asyncio.Task(self.talk_to_websocket(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_websocket_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()
+ 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)
ToastFreeware / toast / stream2beamer.git / blobdiff