Dateien nach „slixmpp/plugins/xep_0293“ hochladen

slixmpp/plugins/xep_0293/__init__.py

@@ -0,0 +1,11 @@
+# slixmpp: The Slick XMPP Library
+# Copyright (C) 2020 Emmanuel Gil Peyrot
+# This file is part of slixmpp.
+# See the file LICENSE for copying permission.
+
+from slixmpp.plugins.base import register_plugin
+
+from slixmpp.plugins.xep_0293.stanza import RtcpFB, Parameter, RtcpFBTrrInt
+from slixmpp.plugins.xep_0293.rtcpfb import XEP_0293
+
+register_plugin(XEP_0293)

slixmpp/plugins/xep_0293/rtcpfb.py

@@ -0,0 +1,60 @@
+# Copyright (C) 2020 Emmanuel Gil Peyrot
+# This file is part of slixmpp.
+# See the file LICENSE for copying permission.
+
+import logging
+
+from typing import Iterable, Tuple, Optional
+
+from slixmpp import JID, Iq
+from slixmpp.plugins import BasePlugin
+from slixmpp.xmlstream import register_stanza_plugin
+from slixmpp.xmlstream.handler import Callback
+from slixmpp.xmlstream.matcher import StanzaPath
+from slixmpp.plugins.xep_0166 import Jingle, Content
+from slixmpp.plugins.xep_0167 import Description, PayloadType
+from slixmpp.plugins.xep_0293 import stanza, RtcpFB, Parameter, RtcpFBTrrInt
+
+import re
+
+log = logging.getLogger(__name__)
+
+
+class XEP_0293(BasePlugin):
+
+    name = 'xep_0293'
+    description = 'XEP-0293: Jingle'
+    dependencies = set(['xep_0166','xep_0167'])
+    stanza = stanza
+
+    def plugin_init(self):
+        register_stanza_plugin(Iq,Jingle)
+        register_stanza_plugin(Jingle,Content,True)
+        register_stanza_plugin(Content,Description)
+        register_stanza_plugin(Description,RtcpFB,True)
+        register_stanza_plugin(Description,PayloadType,True)
+        register_stanza_plugin(PayloadType,RtcpFB,True)
+        register_stanza_plugin(RtcpFB,Parameter,True)
+        register_stanza_plugin(PayloadType,RtcpFBTrrInt,True)
+
+    def make_rtcpfb(self,sdp,id='*'):
+        ret = []
+        iter = re.finditer(r'^a=rtcp-fb:'+id+r' +(\S+)(?: +(\S+))?\r$',sdp,re.M)
+        for m in iter:
+            rtcpfb = RtcpFB()
+            rtcpfb['type'] = m.group(1)
+            if m.group(2):
+                rtcpfb['subtype'] = m.group(2)
+                m = re.search(r'^a=rtcp-fb:'+id+r' +[\S]+(?: +\S+)? +([^:\r]+:[^,\r]+(?:,[^:\r]+:[^,\r])*)\r$',sdp,re.M)
+                if m:
+                    parameters = m.group(1).split(',')
+                    for p in parameters:
+                        parameter = Parameter()
+                        param = p.split(':')
+                        parameter['name'] = param[1]
+                        if len(param)>1:
+                            parameter['value'] = param[2]
+                        rtcpfb.append(parameter)
+            ret.append(rtcpfb)
+        return ret
+

slixmpp/plugins/xep_0293/stanza.py

@@ -0,0 +1,50 @@
+# slixmpp: The Slick XMPP Library
+# Copyright (C) 2020 Emmanuel Gil Peyrot
+# This file is part of slixmpp.
+# See the file LICENSE for copying permission.
+
+from typing import Iterable, List, Tuple, Optional
+from slixmpp.xmlstream import ElementBase, ET
+
+class RtcpFB(ElementBase):
+    name = 'rtcp-fb'
+    namespace = 'urn:xmpp:jingle:apps:rtp:rtcp-fb:0'
+    plugin_attrib = 'rtcp-fb'
+    interfaces = {'type','subtype'}
+    plugin_multi_attrib = 'rtcp-fbs'
+
+    def getSDP(self,id='*'):
+        if self['type']:
+            if self['subtype']:
+                ret = "\r\na=rtcp-fb:%s %s %s" % (id,self['type'],self['subtype'])
+            else:
+                ret = "\r\na=rtcp-fb:%s %s" % (id,self['type'])
+            tmp = ''
+            for parameter in self['parameters']:
+                tmp += ','+parameter.getSDP()
+            if tmp[1:]:
+                ret += ' '+tmp[1:]
+            return ret
+        return None
+
+class Parameter(ElementBase):
+    name = 'parameter'
+    namespace = 'urn:xmpp:jingle:apps:rtp:rtcp-fb:0'
+    plugin_attrib = 'parameter'
+    interfaces = {'name','value'}
+    plugin_multi_attrib = 'parameters'
+
+    def getSDP(self):
+        parameter = self['name']
+        if self['value']:
+            parameter += ":"+self['value']
+        return parameter
+
+class RtcpFBTrrInt(ElementBase):
+    name = 'rtcp-fb-trr-int'
+    namespace = 'urn:xmpp:jingle:apps:rtp:rtcp-fb:0'
+    plugin_attrib = 'rtcp-fb-trr-int'
+    interfaces = {'value'}
+
+    def getSDP(self,id='*'):
+        return "\r\na=rtcp-fb:%s trr-int %s" % (id,self['value'])