common.jingle

38 - def __init__(self, session, media, transport=None):

39 if transport is None: 40 transport = JingleTransportICEUDP() 41 JingleContent.__init__(self, session, transport) 42 self.media = media 43 self._dtmf_running = False 44 self.farsight_media = {'audio': farsight.MEDIA_TYPE_AUDIO, 45 'video': farsight.MEDIA_TYPE_VIDEO}[media] 46 47 self.pipeline = None 48 self.src_bin = None 49 self.stream_failed_once = False 50 51 self.candidates_ready = False # True when local candidates are prepared 52 53 self.callbacks['session-initiate'] += [self.__on_remote_codecs] 54 self.callbacks['content-add'] += [self.__on_remote_codecs] 55 self.callbacks['description-info'] += [self.__on_remote_codecs] 56 self.callbacks['content-accept'] += [self.__on_remote_codecs, 57 self.__on_content_accept] 58 self.callbacks['session-accept'] += [self.__on_remote_codecs, 59 self.__on_content_accept] 60 self.callbacks['session-accept-sent'] += [self.__on_content_accept] 61 self.callbacks['content-accept-sent'] += [self.__on_content_accept] 62 self.callbacks['session-terminate'] += [self.__stop] 63 self.callbacks['session-terminate-sent'] += [self.__stop]

64

65 - def setup_stream(self):

66 # pipeline and bus 67 self.pipeline = gst.Pipeline() 68 bus = self.pipeline.get_bus() 69 bus.add_signal_watch() 70 bus.connect('message', self._on_gst_message) 71 72 # conference 73 self.conference = gst.element_factory_make('fsrtpconference') 74 self.conference.set_property('sdes-cname', self.session.ourjid) 75 self.pipeline.add(self.conference) 76 self.funnel = None 77 78 self.p2psession = self.conference.new_session(self.farsight_media) 79 80 participant = self.conference.new_participant(self.session.peerjid) 81 # FIXME: Consider a workaround, here... 82 # pidgin and telepathy-gabble don't follow the XEP, and it won't work 83 # due to bad controlling-mode 84 params = {'controlling-mode': self.session.weinitiate, 'debug': False} 85 if gajim.config.get('use_stun_server'): 86 stun_server = gajim.config.get('stun_server') 87 if not stun_server and self.session.connection._stun_servers: 88 stun_server = self.session.connection._stun_servers[0]['host'] 89 if stun_server: 90 try: 91 ip = socket.getaddrinfo(stun_server, 0, socket.AF_UNSPEC, 92 socket.SOCK_STREAM)[0][4][0] 93 except socket.gaierror, (errnum, errstr): 94 log.warn('Lookup of stun ip failed: %s' % errstr) 95 else: 96 params['stun-ip'] = ip 97 98 self.p2pstream = self.p2psession.new_stream(participant, 99 farsight.DIRECTION_RECV, 'nice', params)

100

101 - def is_ready(self):

102 return (JingleContent.is_ready(self) and self.candidates_ready)

103

104 - def make_bin_from_config(self, config_key, pipeline, text):

105 pipeline = pipeline % gajim.config.get(config_key) 106 try: 107 bin = gst.parse_bin_from_description(pipeline, True) 108 return bin 109 except GError, error_str: 110 self.session.connection.dispatch('ERROR', 111 (_("%s configuration error") % text.capitalize(), 112 _("Couldn't setup %s. Check your configuration.\n\n" 113 "Pipeline was:\n%s\n\n" 114 "Error was:\n%s") % (text, pipeline, error_str))) 115 raise JingleContentSetupException

116

117 - def add_remote_candidates(self, candidates):

118 JingleContent.add_remote_candidates(self, candidates) 119 # FIXME: connectivity should not be etablished yet 120 # Instead, it should be etablished after session-accept! 121 if self.sent: 122 self.p2pstream.set_remote_candidates(candidates)

123

124 - def batch_dtmf(self, events):

125 """ 126 Send several DTMF tones 127 """ 128 if self._dtmf_running: 129 raise Exception("There is a DTMF batch already running") 130 events = deque(events) 131 self._dtmf_running = True 132 self._start_dtmf(events.popleft()) 133 gobject.timeout_add(500, self._next_dtmf, events)

134

135 - def _next_dtmf(self, events):

136 self._stop_dtmf() 137 if events: 138 self._start_dtmf(events.popleft()) 139 gobject.timeout_add(500, self._next_dtmf, events) 140 else: 141 self._dtmf_running = False

142

143 - def _start_dtmf(self, event):

144 if event in ('*', '#'): 145 event = {'*': farsight.DTMF_EVENT_STAR, 146 '#': farsight.DTMF_EVENT_POUND}[event] 147 else: 148 event = int(event) 149 self.p2psession.start_telephony_event(event, 2, 150 farsight.DTMF_METHOD_RTP_RFC4733)

151

152 - def _stop_dtmf(self):

153 self.p2psession.stop_telephony_event(farsight.DTMF_METHOD_RTP_RFC4733)

154

155 - def _fill_content(self, content):

156 content.addChild(xmpp.NS_JINGLE_RTP + ' description', 157 attrs={'media': self.media}, payload=self.iter_codecs())

158

159 - def _setup_funnel(self):

160 self.funnel = gst.element_factory_make('fsfunnel') 161 self.pipeline.add(self.funnel) 162 self.funnel.set_state(gst.STATE_PLAYING) 163 self.sink.set_state(gst.STATE_PLAYING) 164 self.funnel.link(self.sink)

165

166 - def _on_src_pad_added(self, stream, pad, codec):

167 if not self.funnel: 168 self._setup_funnel() 169 pad.link(self.funnel.get_pad('sink%d'))

170

171 - def _on_gst_message(self, bus, message):

172 if message.type == gst.MESSAGE_ELEMENT: 173 name = message.structure.get_name() 174 log.debug('gst element message: %s: %s' % (name, message)) 175 if name == 'farsight-new-active-candidate-pair': 176 pass 177 elif name == 'farsight-recv-codecs-changed': 178 pass 179 elif name == 'farsight-codecs-changed': 180 if self.sent and self.p2psession.get_property('codecs-ready'): 181 self.send_description_info() 182 elif name == 'farsight-local-candidates-prepared': 183 self.candidates_ready = True 184 if self.is_ready(): 185 self.session.on_session_state_changed(self) 186 elif name == 'farsight-new-local-candidate': 187 candidate = message.structure['candidate'] 188 self.transport.candidates.append(candidate) 189 if self.sent: 190 # FIXME: Is this case even possible? 191 self.send_candidate(candidate) 192 elif name == 'farsight-component-state-changed': 193 state = message.structure['state'] 194 if state == farsight.STREAM_STATE_FAILED: 195 reason = xmpp.Node('reason') 196 reason.setTag('failed-transport') 197 self.session.remove_content(self.creator, self.name, reason) 198 elif name == 'farsight-error': 199 log.error('Farsight error #%d!\nMessage: %s\nDebug: %s' % ( 200 message.structure['error-no'], 201 message.structure['error-msg'], 202 message.structure['debug-msg'])) 203 elif message.type == gst.MESSAGE_ERROR: 204 # TODO: Fix it to fallback to videotestsrc anytime an error occur, 205 # or raise an error, Jingle way 206 # or maybe one-sided stream? 207 if not self.stream_failed_once: 208 self.session.connection.dispatch('ERROR', 209 (_("GStreamer error"), 210 _("Error: %s\nDebug: %s" % (message.structure['gerror'], 211 message.structure['debug'])))) 212 213 sink_pad = self.p2psession.get_property('sink-pad') 214 215 # Remove old source 216 self.src_bin.get_pad('src').unlink(sink_pad) 217 self.src_bin.set_state(gst.STATE_NULL) 218 self.pipeline.remove(self.src_bin) 219 220 if not self.stream_failed_once: 221 # Add fallback source 222 self.src_bin = self.get_fallback_src() 223 self.pipeline.add(self.src_bin) 224 self.src_bin.get_pad('src').link(sink_pad) 225 self.stream_failed_once = True 226 else: 227 reason = xmpp.Node('reason') 228 reason.setTag('failed-application') 229 self.session.remove_content(self.creator, self.name, reason) 230 231 # Start playing again 232 self.pipeline.set_state(gst.STATE_PLAYING)

233

234 - def get_fallback_src(self):

235 return gst.element_factory_make('fakesrc')

236

237 - def __on_content_accept(self, stanza, content, error, action):

238 if self.accepted: 239 if self.transport.remote_candidates: 240 self.p2pstream.set_remote_candidates(self.transport.remote_candidates) 241 self.transport.remote_candidates = [] 242 # TODO: farsight.DIRECTION_BOTH only if senders='both' 243 self.p2pstream.set_property('direction', farsight.DIRECTION_BOTH) 244 self.on_negotiated()

245

246 - def __on_remote_codecs(self, stanza, content, error, action):

247 """ 248 Get peer codecs from what we get from peer 249 """ 250 251 codecs = [] 252 for codec in content.getTag('description').iterTags('payload-type'): 253 c = farsight.Codec(int(codec['id']), codec['name'], 254 self.farsight_media, int(codec['clockrate'])) 255 if 'channels' in codec: 256 c.channels = int(codec['channels']) 257 else: 258 c.channels = 1 259 c.optional_params = [(str(p['name']), str(p['value'])) for p in \ 260 codec.iterTags('parameter')] 261 codecs.append(c) 262 263 if codecs: 264 # FIXME: Handle this case: 265 # glib.GError: There was no intersection between the remote codecs and 266 # the local ones 267 self.p2pstream.set_remote_codecs(codecs)

268

269 - def iter_codecs(self):

270 codecs = self.p2psession.get_property('codecs') 271 for codec in codecs: 272 attrs = {'name': codec.encoding_name, 273 'id': codec.id, 274 'channels': codec.channels} 275 if codec.clock_rate: 276 attrs['clockrate'] = codec.clock_rate 277 if codec.optional_params: 278 payload = (xmpp.Node('parameter', {'name': name, 'value': value}) 279 for name, value in codec.optional_params) 280 else: 281 payload = () 282 yield xmpp.Node('payload-type', attrs, payload)

283

284 - def __stop(self, *things):

285 self.pipeline.set_state(gst.STATE_NULL)

286

287 - def __del__(self):

288 self.__stop()

289

290 - def destroy(self):

291 JingleContent.destroy(self) 292 self.p2pstream.disconnect_by_func(self._on_src_pad_added) 293 self.pipeline.get_bus().disconnect_by_func(self._on_gst_message)

Source Code for Module common.jingle_rtp