Author: liori Date: 2007-08-19 21:07:49 +0200 (Sun, 19 Aug 2007) New Revision: 8506
Added: branches/jingle/src/common/jingle-handler.py branches/jingle/src/common/jingle.popen.py Log: Jingle: a try with popen()ing external file for p2p connection
Added: branches/jingle/src/common/jingle-handler.py =================================================================== --- branches/jingle/src/common/jingle-handler.py (rev 0) +++ branches/jingle/src/common/jingle-handler.py 2007-08-19 19:07:49 UTC (rev 8506) @@ -0,0 +1,258 @@ +#!/usr/bin/python + +import sys, dl, gst, gobject +sys.setdlopenflags(dl.RTLD_NOW | dl.RTLD_GLOBAL) +import farsight +from socket import * + +TYPE_CANDIDATE=0 +TYPE_CODEC=1 +TYPE_READY_BYTE=2 + +MODE_DUPLEX=0 +MODE_AUDIO=1 +MODE_VIDEO=2 + +send_sock = 0 +recv_sock = 0 +serv_addr = None +audio_stream = None +video_stream = None + +def stream_error(stream, error, debug): + pass + #print "stream_error: stream=%r error=%s" % (stream, error) + +def session_error(session, error, debug): + pass + #print "session_error: session=%r error=%s" % (session, error) + +def new_active_candidate_pair(stream, native, remote): + pass + #print "new_active_candidate_pair: native %s, remote %s" % (native, remote) + #send_codecs(stream.get_local_codecs()) + +def codec_changed(stream, codec_id): + #print "codec_changed: codec_id=%d, stream=%r" % (codec_id, stream) + pass + +def native_candidates_prepared(stream): + return + print "native_candidates_prepared: stream=%r" % stream + for info in stream.get_native_candidate_list(): + print "Local transport candidate: %s %d %s %s %s %d pref %f" % ( + info['candidate_id'], info['component'], + (info['proto']==farsight.NETWORK_PROTOCOL_TCP) and "TCP" or "UDP", + info['proto_subtype'], info['ip'], info['port'], info['preference']); + +def show_element(e): + return + print 'element: ...' + +def state_changed(stream, state, dir): + if state==farsight.STREAM_STATE_DISCONNECTED: + #print "state_changed: disconnected" + pass + elif state==farsight.STREAM_STATE_CONNECTING: + #print "state_changed: connecting" + pass + elif state==farsight.STREAM_STATE_CONNECTED: + #print "state_changed: connectied" + #print "WW: stream.signal_native_candidates_prepared()" + #print "WW: stream.start()" + stream.signal_native_candidates_prepared() + stream.start() + +def send_codecs(codecs): + global send_sock, serv_addr + for codec in codecs: + if 'channels' not in codec: codec['channels']=1 + s="%d %d %s %d %d %d " % (TYPE_CODEC, codec['media_type'], + codec['encoding_name'], codec['id'], codec['clock_rate'], + codec['channels']) + print s + s="%d %d %s %d %d %d " % (TYPE_CODEC, codec['media_type'], 'LAST', 0, 0, 0) + print s + +def send_candidate(type, trans): + global send_sock, serv_addr + s="%d %d %s %s %d %s %s" % (TYPE_CANDIDATE, + type, trans['candidate_id'], trans['ip'], trans['port'], + trans['username'], trans['password']) + print s + +def new_native_candidate(stream, candidate_id): + candidate=stream.get_native_candidate(candidate_id) + trans=candidate[0] + + #print "New native candidate: <id: %s, component: %d, ip: %s port: %d proto: %d, proto_subtype: %s, proto_profile: %s, preference: %f, type: %d, username: %s, password: %s>" % ( + # trans['candidate_id'], trans['component'], trans['ip'], trans['port'], trans['proto'], + # trans['proto_subtype'], trans['proto_profile'], trans['preference'], trans['type'], + # trans['username'], trans['password']) + + type=stream.get_property("media-type") + + send_candidate(type, trans) + +remote_codecs_audio = [] +remote_codecs_video = [] +def add_remote_codec(stream, pt, encoding_name, media_type, clock_rate, channels): + global remote_codecs_audio, remote_codecs_video + if encoding_name=="LAST": + if media_type==farsight.MEDIA_TYPE_AUDIO: + #print "WW: set_remote_codecs(remote_codecs_audio), %r" % (remote_codecs_audio,) + stream.set_remote_codecs(remote_codecs_audio) + else: + stream.set_remote_codecs(remote_codecs_video) + else: + codec={'id': pt, 'encoding_name': encoding_name, 'media_type': media_type, + 'clock_rate': clock_rate, 'channels': channels} + if media_type==farsight.MEDIA_TYPE_AUDIO: + remote_codecs_audio.append(codec) + elif media_type==farsight.MEDIA_TYPE_VIDEO: + remote_codecs_video.append(codec) + + #for codec in remote_codecs_audio: + # print "added audio codec: %d: %s/%d found" % (codec['id'], codec['encoding_name'], codec['clock_rate']) + #for codec in remote_codecs_video: + # print "added video codec: %d: %s/%d found" % (codec['id'], codec['encoding_name'], codec['clock_rate']) + +def add_remote_candidate(stream, id, ip, port, username, password): + if not stream: return + + trans={'candidate_id': id, 'component': 1, 'ip': ip, 'port': port, 'proto': farsight.NETWORK_PROTOCOL_UDP, + 'proto_subtype': 'RTP', 'proto_profile': 'AVP', 'preference': 1.0, + 'type': farsight.CANDIDATE_TYPE_LOCAL, 'username': username, 'password': password} + + #print "WW: add_remote_candidate(%r)" % ([trans],) + stream.add_remote_candidate([trans]); + +def receive_loop(ch, cond, *data): + #print 'receive_loop called!' + global recv_sock, serv_addr, audio_stream, remote_stream + #print "waiting for msg from %r" % (serv_addr,) + buf=raw_input() + #buf,a = recv_sock.recvfrom(1500) + #print "got message! ",buf + msg = buf.split(' ') + type = int(msg.pop(0)) + #print msg + if type==TYPE_CANDIDATE: + media_type, id, ip, port, username, password = msg[:6] + media_type=int(media_type) + port=int(port) + #print "Received %d %s %s %d %s %s" % (media_type, id, ip, port, username, password) + if media_type==farsight.MEDIA_TYPE_AUDIO: + add_remote_candidate(audio_stream, id, ip, port, username, password) + elif media_type==farsight.MEDIA_TYPE_VIDEO: + add_remote_candidate(video_stream, id, ip, port, username, password) + elif type==TYPE_CODEC: + media_type, encoding_name, pt, clock_rate, channels=msg[:5] + media_type=int(media_type) + pt=int(pt) + clock_rate=int(clock_rate) + channels=int(channels) + + #print "Received %d %s %d %d %d" % (media_type, encoding_name, pt, clock_rate, channels) + + if media_type==farsight.MEDIA_TYPE_AUDIO: + add_remote_codec(audio_stream, pt, encoding_name, media_type, clock_rate, channels) + elif media_type==farsight.MEDIA_TYPE_VIDEO: + add_remote_codec(video_stream, pt, encoding_name, media_type, clock_rate, channels) + elif type==TYPE_READY_BYTE: + #print "got ready byte" + pass + return True + +def set_xoverlay(bus, message, xid): + print "set_xoverlay: called" + if message!=gst.MESSAGE_ELEMENT: return gst.BUS_PASS + + if not message.structure.has_name('prepare-xwindow-id'): return gst.BUS_PASS + + print "set_xoverlay: setting x overlay window id" + message.set_xwindow_id(xid) + + return gst.BUS_DROP + +def main(): + global audio_stream, remote_stream + #if len(sys.argv)<4 or len(sys.argv)>6: + # #print 'usage: %s remoteip remoteport localport [mode] [xid]' % sys.argv[0] + # return -1 + + #if len(sys.argv)>=5: + # mode = int(sys.argv[4]) + #else: + mode = MODE_DUPLEX + + #setup_send(sys.argv[1], int(sys.argv[2])) + #print "Sending to %s %d listening on port %d" % (sys.argv[1], int(sys.argv[2]), int(sys.argv[3])) + + #recv_chan=setup_recv(int(sys.argv[3])) + + #do_handshake(recv_chan) + + gobject.io_add_watch(0, gobject.IO_IN, receive_loop) + + session = setup_rtp_session() + + if mode==MODE_AUDIO or mode==MODE_VIDEO: + audio_stream = setup_rtp_stream(session, farsight.MEDIA_TYPE_AUDIO) + if mode==MODE_VIDEO or mode==MODE_DUPLEX: + video_stream = setup_rtp_stream(session, farsight.MEDIA_TYPE_VIDEO) + + if audio_stream: + audio_stream.set_active_codec(8) + + alsasrc = gst.element_factory_make('audiotestsrc', 'src') + alsasink= gst.element_factory_make('alsasink', 'alsasink') + + alsasink.set_property('sync', False) + alsasink.set_property('latency-time', 20000) + alsasink.set_property('buffer-time', 80000) + alsasrc.set_property('blocksize', 320) + #alsasrc.set_property('latency-time', 20000) + alsasrc.set_property('is-live', True) + + #print "WW: set_sink(%r)" % (alsasink,) + audio_stream.set_sink(alsasink) + #print "WW: set_source(%r)" % (alsasrc,) + audio_stream.set_source(alsasrc) + + #if video_stream: assert False + + gobject.MainLoop().run() + #print 'bu!' + +def setup_rtp_session(): + session = farsight.farsight_session_factory_make('rtp') + + session.connect('error', session_error) + return session + +def setup_rtp_stream(session, type): + stream = session.create_stream(type, farsight.STREAM_DIRECTION_BOTH) + + stream.set_property('transmitter', 'libjingle') + + stream.connect('error', stream_error) + stream.connect('new-active-candidate-pair', new_active_candidate_pair) + stream.connect('codec-changed', codec_changed) + stream.connect('native-candidates-prepared', native_candidates_prepared) + stream.connect('state-changed', state_changed) + stream.connect('new-native-candidate', new_native_candidate) + + possible_codecs = stream.get_local_codecs() + + #for codec in possible_codecs: + # print "codec: %d: %s/%d found" % (codec['id'], codec['encoding_name'], codec['clock_rate']) + + send_codecs(possible_codecs) + + stream.prepare_transports() + + return stream + + +main()
Property changes on: branches/jingle/src/common/jingle-handler.py ___________________________________________________________________ Name: svn:executable + *
Copied: branches/jingle/src/common/jingle.popen.py (from rev 8498, branches/jingle/src/common/jingle.py) =================================================================== --- branches/jingle/src/common/jingle.popen.py (rev 0) +++ branches/jingle/src/common/jingle.popen.py 2007-08-19 19:07:49 UTC (rev 8506) @@ -0,0 +1,666 @@ +## +## Copyright (C) 2006 Gajim Team +## +## This program is free software; you can redistribute it and/or modify +## it under the terms of the GNU General Public License as published +## by the Free Software Foundation; version 2 only. +## +## This program is distributed in the hope that it will be useful, +## but WITHOUT ANY WARRANTY; without even the implied warranty of +## MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +## GNU General Public License for more details. +## +''' Handles the jingle signalling protocol. ''' + +import gajim +import xmpp +import gobject + +import sys, dl, gst, gobject +sys.setdlopenflags(dl.RTLD_NOW | dl.RTLD_GLOBAL) +import farsight + + +import meta + +class JingleStates(object): + ''' States in which jingle session may exist. ''' + ended=0 + pending=1 + active=2 + +class Exception(object): pass +class WrongState(Exception): pass +class NoCommonCodec(Exception): pass + +class JingleSession(object): + ''' This represents one jingle session. ''' + __metaclass__=meta.VerboseClassType + def __init__(self, con, weinitiate, jid, sid=None): + ''' con -- connection object, + weinitiate -- boolean, are we the initiator? + jid - jid of the other entity''' + self.contents={} # negotiated contents + self.connection=con # connection to use + # our full jid + self.ourjid=gajim.get_jid_from_account(self.connection.name)+'/'+con.server_resource + self.peerjid=jid # jid we connect to + # jid we use as the initiator + self.initiator=weinitiate and self.ourjid or self.peerjid + # jid we use as the responder + self.responder=weinitiate and self.peerjid or self.ourjid + # are we an initiator? + self.weinitiate=weinitiate + # what state is session in? (one from JingleStates) + self.state=JingleStates.ended + if not sid: + sid=con.connection.getAnID() + self.sid=sid # sessionid + + # callbacks to call on proper contents + # use .prepend() to add new callbacks, especially when you're going + # to send error instead of ack + self.callbacks={ + 'content-accept': [self.__contentAcceptCB, self.__broadcastCB, self.__defaultCB], + 'content-add': [self.__defaultCB], + 'content-modify': [self.__defaultCB], + 'content-remove': [self.__defaultCB], + 'session-accept': [self.__contentAcceptCB, self.__broadcastCB, self.__defaultCB], + 'session-info': [self.__defaultCB], + 'session-initiate': [self.__sessionInitiateCB, self.__broadcastCB, self.__defaultCB], + 'session-terminate': [self.__defaultCB], + 'transport-info': [self.__broadcastCB, self.__defaultCB], + 'iq-result': [], + 'iq-error': [], + } + + ''' Middle-level functions to manage contents. Handle local content + cache and send change notifications. ''' + def addContent(self, name, content, creator='we'): + ''' Add new content to session. If the session is active, + this will send proper stanza to update session. + The protocol prohibits changing that when pending. + Creator must be one of ('we', 'peer', 'initiator', 'responder')''' + if self.state==JingleStates.pending: + raise WrongState + + if (creator=='we' and self.weinitiate) or (creator=='peer' and not self.weinitiate): + creator='initiator' + elif (creator=='peer' and self.weinitiate) or (creator=='we' and not self.weinitiate): + creator='responder' + content.creator = creator + content.name = name + self.contents[(creator,name)]=content + + if self.state==JingleStates.active: + pass # TODO: send proper stanza, shouldn't be needed now + + def removeContent(self, creator, name): + ''' We do not need this now ''' + pass + + def modifyContent(self, creator, name, *someother): + ''' We do not need this now ''' + pass + + ''' Middle-level function to do stanza exchange. ''' + def startSession(self): + ''' Start session. ''' + self.__sessionInitiate() + + def sendSessionInfo(self): pass + + ''' Callbacks. ''' + def stanzaCB(self, stanza): + ''' A callback for ConnectionJingle. It gets stanza, then + tries to send it to all internally registered callbacks. + First one to raise xmpp.NodeProcessed breaks function.''' + jingle = stanza.getTag('jingle') + error = stanza.getTag('error') + if error: + # it's an iq-error stanza + action = 'iq-error' + elif jingle: + # it's a jingle action + action = jingle.getAttr('action') + else: + # it's an iq-result (ack) stanza + action = 'iq-result' + + callables = self.callbacks[action] + + try: + for callable in callables: + callable(stanza=stanza, jingle=jingle, error=error, action=action) + except xmpp.NodeProcessed: + pass + + def __defaultCB(self, stanza, jingle, error, action): + ''' Default callback for action stanzas -- simple ack + and stop processing. ''' + response = stanza.buildReply('result') + self.connection.connection.send(response) + + def __contentAcceptCB(self, stanza, jingle, error, action): + ''' Called when we get content-accept stanza or equivalent one + (like session-accept).''' + # check which contents are accepted + for content in jingle.iterTags('content'): + creator = content['creator'] + name = content['name'] + + def __sessionInitiateCB(self, stanza, jingle, error, action): + ''' We got a jingle session request from other entity, + therefore we are the receiver... Unpack the data. ''' + self.initiator = jingle['initiator'] + self.responder = self.ourjid + self.peerjid = self.initiator + + fail = True + for element in jingle.iterTags('content'): + # checking what kind of session this will be + desc_ns = element.getTag('description').getNamespace() + tran_ns = element.getTag('transport').getNamespace() + if desc_ns==xmpp.NS_JINGLE_AUDIO and tran_ns==xmpp.NS_JINGLE_ICE_UDP: + # we've got voip content + self.addContent(element['name'], JingleVoiP(self), 'peer') + fail = False + + if fail: + # TODO: we should send <unsupported-content/> inside too + self.connection.connection.send( + xmpp.Error(stanza, xmpp.NS_STANZAS + 'feature-not-implemented')) + self.connection.deleteJingle(self) + raise xmpp.NodeProcessed + + self.state = JingleStates.pending + + def __broadcastCB(self, stanza, jingle, error, action): + ''' Broadcast the stanza contents to proper content handlers. ''' + for content in jingle.iterTags('content'): + name = content['name'] + creator = content['creator'] + cn = self.contents[(creator, name)] + cn.stanzaCB(stanza, content, error, action) + + def on_p2psession_error(self, *anything): + print self.weinitiate, "Farsight session error!" + + ''' Methods that make/send proper pieces of XML. They check if the session + is in appropriate state. ''' + def __makeJingle(self, action): + stanza = xmpp.Iq(typ='set', to=xmpp.JID(self.peerjid)) + jingle = stanza.addChild('jingle', attrs={ + 'xmlns': 'http://www.xmpp.org/extensions/xep-0166.html#ns', + 'action': action, + 'initiator': self.initiator, + 'responder': self.responder, + 'sid': self.sid}) + return stanza, jingle + + def __appendContent(self, jingle, content, full=True): + ''' Append <content/> element to <jingle/> element, + with (full=True) or without (full=False) <content/> + children. ''' + if full: + jingle.addChild(node=content.toXML()) + else: + jingle.addChild('content', + attrs={'name': content.name, 'creator': content.creator}) + + def __appendContents(self, jingle, full=True): + ''' Append all <content/> elements to <jingle/>.''' + # TODO: integrate with __appendContent? + # TODO: parameters 'name', 'content'? + for content in self.contents.values(): + self.__appendContent(jingle, content, full=full) + + def __sessionInitiate(self): + assert self.state==JingleStates.ended + stanza, jingle = self.__makeJingle('session-initiate') + self.__appendContents(jingle) + self.connection.connection.send(stanza) + + def __sessionAccept(self): + assert self.state==JingleStates.pending + stanza, jingle = self.__jingle('session-accept') + self.__appendContents(jingle, False) + self.connection.connection.send(stanza) + self.state=JingleStates.active + + def __sessionInfo(self, payload=None): + assert self.state!=JingleStates.ended + stanza, jingle = self.__jingle('session-info') + if payload: + jingle.addChild(node=payload) + self.connection.connection.send(stanza) + + def __sessionTerminate(self): + assert self.state!=JingleStates.ended + stanza, jingle = self.__jingle('session-terminate') + self.connection.connection.send(stanza) + + def __contentAdd(self): + assert self.state==JingleStates.active + + def __contentAccept(self): + assert self.state!=JingleStates.ended + + def __contentModify(self): + assert self.state!=JingleStates.ended + + def __contentRemove(self): + assert self.state!=JingleStates.ended + + def sendTransportInfo(self, content): + assert self.state!=JingleStates.ended + stanza, jingle = self.__makeJingle('transport-info') + jingle.addChild(node=content) + self.connection.connection.send(stanza) + + '''Callbacks''' + def sessionTerminateCB(self, stanza): pass + +class Codec(object): + ''' This class keeps description of a single codec. ''' + def __init__(self, name, id=None, **params): + ''' Create new codec description. ''' + self.name = name + self.id = id + self.attrs = {'name': self.name, 'id': self.id, 'channels': 1} + for key in ('channels', 'clockrate', 'maxptime', 'ptime'): + if key in params: + self.attrs[key]=params[key] + del params[key] + self.params = params + + def __eq__(a, b): + ''' Compare two codec descriptions. ''' + # TODO: check out what should be tested... + if a.name!=b.name: return False + # ... + return True + + def toXML(self): + return xmpp.Node('payload-type', + attrs=self.attrs, + payload=(xmpp.Node('parameter', {'name': k, 'value': v}) for k,v in self.params)) + +class JingleAudioSession(object): +# __metaclass__=meta.VerboseClassType + def __init__(self, content, fromNode): + self.content = content + + self.initiator_codecs=[] + self.responder_codecs=[] + + if fromNode: + # read all codecs peer understand + for payload in fromNode.iterTags('payload-type'): + attrs = fromNode.getAttrs().copy() + for param in fromNode.iterTags('parameter'): + attrs[param['name']]=param['value'] + self.initiator_codecs.append(Codec(**attrs)) + + def sessionInitiateCB(self, stanza, ourcontent): + pass + + ''' "Negotiation" of codecs... simply presenting what *we* can do, nothing more... ''' + def getOurCodecs(self, other=None): + ''' Get a list of codecs we support. Try to get them in the same + order as the codecs of our peer. If other!=None, raise + a NoCommonCodec error if no codecs both sides support (None means + we are initiating the connection and we don't know the other + peer's codecs.) ''' + # for now we "understand" only one codec -- speex with clockrate 16000 + # so we have an easy job to do... (codecs sorted in order of preference) + supported_codecs=[ + Codec('speex', clockrate='16000'), + ] + + other_l = other if other is not None else [] + our_l = supported_codecs[:] + out = [] + ids = range(128) + for codec in other_l: + if codec in our_l: + out.append(codec) + our_l.remove(codec) + try: ids.remove(codec.id) + except ValueError: pass # when id is not a dynamic one + + if other is not None and len(out)==0: + raise NoCommonCodec + + for codec in our_l: + if not codec.id or codec.id not in ids: + codec.id = ids.pop() + out.append(codec) + + return out + + ''' Methods for making proper pieces of XML. ''' + def __codecsList(self, codecs): + ''' Prepares a description element with codecs given as a parameter. ''' + return xmpp.Node(xmpp.NS_JINGLE_AUDIO+' description', + payload=(codec.toXML() for codec in codecs)) + + def toXML(self): + if not self.initiator_codecs: + # we are the initiator, so just send our codecs + self.initiator_codecs = self.getOurCodecs() + return self.__codecsList(self.initiator_codecs) + else: + # we are the responder, we SHOULD adjust our codec list + self.responder_codecs = self.getOurCodecs(self.initiator_codecs) + return self.__codecsList(self.responder_codecs) + +class JingleContent(object): + ''' An abstraction of content in Jingle sessions. ''' + def __init__(self, session, node=None): + self.session = session + # will be filled by JingleSession.add_content() + # don't uncomment these lines, we will catch more buggy code then + # (a JingleContent not added to session shouldn't send anything) + #self.creator = None + #self.name = None + +class JingleVoiP(JingleContent): + ''' Jingle VoiP sessions consist of audio content transported + over an ICE UDP protocol. ''' +# __metaclass__=meta.VerboseClassType + def __init__(self, session, node=None): + JingleContent.__init__(self, session, node) + self.got_codecs = False + self.codecs = [] + + #if node is None: + # self.audio = JingleAudioSession(self) + #else: + # self.audio = JingleAudioSession(self, node.getTag('content')) + #self.transport = JingleICEUDPSession(self) + self.setupStream() + + def stanzaCB(self, stanza, content, error, action): + ''' Called when something related to our content was sent by peer. ''' + callbacks = { + 'content-accept': [self.__getRemoteCodecsCB], + 'content-add': [], + 'content-modify': [], + 'content-remove': [], + 'session-accept': [self.__getRemoteCodecsCB], + 'session-info': [], + 'session-initiate': [self.__getRemoteCodecsCB], + 'session-terminate': [], + 'transport-info': [self.__transportInfoCB], + 'iq-result': [], + 'iq-error': [], + }[action] + for callback in callbacks: + callback(stanza, content, error, action) + + def __getRemoteCodecsCB(self, stanza, content, error, action): + if self.got_codecs: return + + codecs = [] + for codec in content.getTag('description').iterTags('payload-type'): + c = {'id': int(codec['id']), + 'encoding_name': codec['name'], + 'media_type': farsight.MEDIA_TYPE_AUDIO, + 'channels': 1, + 'params': dict((p['name'], p['value']) for p in codec.iterTags('parameter'))} + if 'channels' in codec: c['channels']=codec['channels'] + codecs.append(c) + self.p2pin.write('%d %d %s %d %d %d\n' % (1, farsight.MEDIA_TYPE_AUDIO, + codec['encoding_name'], codec['id'], codec['clock_rate'], codec['channels'])) + if len(codecs)==0: return + + print self.session.weinitiate, "#farsight_stream_set_remote_codecs" + #self.p2pstream.set_remote_codecs(codecs) + self.p2pin.write('%d %d %s %d %d %d\n' % (1, farsight.MEDIA_TYPE_AUDIO, + 'LAST', 0, 0, 0)) + self.got_codecs=True + + def __transportInfoCB(self, stanza, content, error, action): + ''' Got a new transport candidate. ''' + candidates = [] + for candidate in content.getTag('transport').iterTags('candidate'): + cand={ + # 'candidate_id': str(self.session.connection.connection.getAnID()), + 'candidate_id': candidate['cid'], + 'component': int(candidate['component']), + 'ip': candidate['ip'], + 'port': int(candidate['port']), + 'proto': candidate['protocol']=='udp' and farsight.NETWORK_PROTOCOL_UDP \ + or farsight.NETWORK_PROTOCOL_TCP, + 'proto_subtype':'RTP', + 'proto_profile':'AVP', + 'preference': float(candidate['priority'])/100000, + # 'type': farsight.CANDIDATE_TYPE_LOCAL, + 'type': int(candidate['type']), + } + if 'ufrag' in candidate: cand['username']=candidate['ufrag'] + if 'pwd' in candidate: cand['password']=candidate['pwd'] + + self.p2pin.write('%d %d %s %s %d %s %s\n' % (0, int(candidate['type']), candidate['cid'], candidate['ip'], int(candidate['port']), candidate['ufrag'], candidate['pwd'])) + candidates.append(cand) + print self.session.weinitiate, "#add_remote_candidate" + #self.p2pstream.add_remote_candidate(candidates) + + def toXML(self): + ''' Return proper XML for <content/> element. ''' + return xmpp.Node('content', + attrs={'name': self.name, 'creator': self.creator, 'profile': 'RTP/AVP'}, + payload=[ + xmpp.Node(xmpp.NS_JINGLE_AUDIO+' description', payload=self.iterCodecs()), + xmpp.Node(xmpp.NS_JINGLE_ICE_UDP+' transport') + ]) + + def __content(self, payload=[]): + ''' Build a XML content-wrapper for our data. ''' + return xmpp.Node('content', + attrs={'name': self.name, 'creator': self.creator, 'profile': 'RTP/AVP'}, + payload=payload) + + def new_data(self, *things): + buf = self.p2pout.readline() + print "RECV: %r"%buf + msg = buf.split(' ') + try: + type = int(msg.pop(0)) + except: + return True + #print msg + if type==0: + media_type, id, ip, port, username, password = msg[:6] + media_type=int(media_type) + port=int(port) + #print "Received %d %s %s %d %s %s" % (media_type, id, ip, port, username, password) + self.send_candidate({'candidate_id': id, 'component': 1, 'ip': ip, + 'port': port, 'username': username, 'password': password, + 'proto': farsight.NETWORK_PROTOCOL_UDP, 'proto_subtype': 'RTP', + 'proto_profile': 'AVP', 'preference': 1.0, 'type': 0}) + elif type==1: + media_type, encoding_name, pt, clock_rate, channels=msg[:5] + media_type=int(media_type) + pt=int(pt) + clock_rate=int(clock_rate) + channels=int(channels) + + #print "Received %d %s %d %d %d" % (media_type, encoding_name, pt, clock_rate, channels) + if encoding_name!='LAST': + self.codecs.append((media_type, encoding_name, pt, clock_rate, channels)) + else: + self.got_codecs=True + return True + + def setupStream(self): + import popen2 + self.p2pout, self.p2pin = popen2.popen2('/home/liori/Projekty/SoC/trunk/src/common/jingle-handler.py') + gobject.io_add_watch(self.p2pout, gobject.IO_IN, self.new_data) + while not self.got_codecs: + self.new_data() + return + print self.session.weinitiate, "#farsight_session_create_stream" + self.p2pstream = self.session.p2psession.create_stream( + farsight.MEDIA_TYPE_AUDIO, farsight.STREAM_DIRECTION_BOTH) + self.p2pstream.set_property('transmitter', 'libjingle') + self.p2pstream.connect('error', self.on_p2pstream_error) + self.p2pstream.connect('new-active-candidate-pair', self.on_p2pstream_new_active_candidate_pair) + self.p2pstream.connect('codec-changed', self.on_p2pstream_codec_changed) + self.p2pstream.connect('native-candidates-prepared', self.on_p2pstream_native_candidates_prepared) + self.p2pstream.connect('state-changed', self.on_p2pstream_state_changed) + self.p2pstream.connect('new-native-candidate', self.on_p2pstream_new_native_candidate) + + self.p2pstream.set_remote_codecs(self.p2pstream.get_local_codecs()) + + print self.session.weinitiate, "#farsight_stream_prepare_transports" + self.p2pstream.prepare_transports() + + print self.session.weinitiate, "#farsight_stream_set_active_codec" + self.p2pstream.set_active_codec(8) #??? + + sink = gst.element_factory_make('alsasink') + sink.set_property('sync', False) + sink.set_property('latency-time', 20000) + sink.set_property('buffer-time', 80000) + + src = gst.element_factory_make('audiotestsrc') + src.set_property('blocksize', 320) + #src.set_property('latency-time', 20000) + src.set_property('is-live', True) + + print self.session.weinitiate, "#farsight_stream_set_sink" + self.p2pstream.set_sink(sink) + print self.session.weinitiate, "#farsight_stream_set_source" + self.p2pstream.set_source(src) + + def on_p2pstream_error(self, *whatever): pass + def on_p2pstream_new_active_candidate_pair(self, stream, native, remote): + print self.session.weinitiate, "##new_active_candidate_pair" + #print "New native candidate pair: %s, %s" % (native, remote) + def on_p2pstream_codec_changed(self, stream, codecid): + print self.session.weinitiate, "##codec_changed" + #print "Codec changed: %d" % codecid + def on_p2pstream_native_candidates_prepared(self, *whatever): + print self.session.weinitiate, "##native_candidates_prepared" + #print "Native candidates prepared: %r" % whatever + for candidate in self.p2pstream.get_native_candidate_list(): + self.send_candidate(candidate) + def on_p2pstream_state_changed(self, stream, state, dir): + print self.session.weinitiate, "##state_changed" + #print "State: %d, Dir: %d" % (state, dir) + if state==farsight.STREAM_STATE_CONNECTED: + print self.session.weinitiate, "#farsight_stream_signal_native_candidates_prepared" + stream.signal_native_candidates_prepared() + print self.session.weinitiate, "#farsight_stream_start" + stream.start() + def on_p2pstream_new_native_candidate(self, p2pstream, candidate_id): + print self.session.weinitiate, "##new_native_candidate" + print self.session.weinitiate, "#get_native_candidate" + candidates = p2pstream.get_native_candidate(candidate_id) + print self.session.weinitiate, "#!", repr(candidates) + + for candidate in candidates: + self.send_candidate(candidate) + def send_candidate(self, candidate): + attrs={ + 'cid': candidate['candidate_id'], + 'component': candidate['component'], + 'foundation': '1', # hack + 'generation': '0', + 'type': candidate['type'], + 'ip': candidate['ip'], + 'network': '0', + 'port': candidate['port'], + 'priority': int(100000*candidate['preference']), # hack + 'protocol': candidate['proto']==farsight.NETWORK_PROTOCOL_UDP and 'udp' or 'tcp', + } + if 'username' in candidate: attrs['ufrag']=candidate['username'] + if 'password' in candidate: attrs['pwd']=candidate['password'] + c=self.__content() + t=c.addChild(xmpp.NS_JINGLE_ICE_UDP+' transport') + t.addChild('candidate', attrs=attrs) + self.session.sendTransportInfo(c) + + def iterCodecs(self): + for codec in self.codecs: + media_type, encoding_name, pt, clock_rate, channels=codec + yield xmpp.Node('payload-type', { + 'name': encoding_name, + 'id': pt, + 'channels': channels, + 'clock_rate': clock_rate}) + +# print self.session.weinitiate, "#farsight_stream_get_local_codecs" +# codecs=self.p2pstream.get_local_codecs() +# for codec in codecs: +# a = {'name': codec['encoding_name'], +# 'id': codec['id'], +# 'channels': 1} +# if 'clock_rate' in codec: a['clockrate']=codec['clock_rate'] +# if 'optional_params' in codec: +# p = (xmpp.Node('parameter', {'name': name, 'value': value}) +# for name, value in codec['optional_params'].iteritems()) +# else: p = () +# yield xmpp.Node('payload-type', a, p) + +class ConnectionJingle(object): + ''' This object depends on that it is a part of Connection class. ''' + def __init__(self): + # dictionary: (jid, sessionid) => JingleSession object + self.__sessions = {} + + # dictionary: (jid, iq stanza id) => JingleSession object, + # one time callbacks + self.__iq_responses = {} + + def addJingle(self, jingle): + ''' Add a jingle session to a jingle stanza dispatcher + jingle - a JingleSession object. + ''' + self.__sessions[(jingle.peerjid, jingle.sid)]=jingle + + def deleteJingle(self, jingle): + ''' Remove a jingle session from a jingle stanza dispatcher ''' + del self.__session[(jingle.peerjid, jingle.sid)] + + def _JingleCB(self, con, stanza): + ''' The jingle stanza dispatcher. + Route jingle stanza to proper JingleSession object, + or create one if it is a new session. + TODO: Also check if the stanza isn't an error stanza, if so + route it adequatelly.''' + + # get data + jid = stanza.getFrom() + id = stanza.getID() + + if (jid, id) in self.__iq_responses.keys(): + self.__iq_responses[(jid, id)].stanzaCB(stanza) + del self.__iq_responses[(jid, id)] + raise xmpp.NodeProcessed + + jingle = stanza.getTag('jingle') + if not jingle: return + sid = jingle.getAttr('sid') + + # do we need to create a new jingle object + if (jid, sid) not in self.__sessions: + # TODO: we should check its type here... + newjingle = JingleSession(con=self, weinitiate=False, jid=jid, sid=sid) + self.addJingle(newjingle) + + # we already have such session in dispatcher... + self.__sessions[(jid, sid)].stanzaCB(stanza) + + raise xmpp.NodeProcessed + + def addJingleIqCallback(self, jid, id, jingle): + self.__iq_responses[(jid, id)]=jingle + + def startVoiP(self, jid): + jingle = JingleSession(self, weinitiate=True, jid=jid) + self.addJingle(jingle) + jingle.addContent('voice', JingleVoiP(jingle)) + jingle.startSession()