Changeset 402

Show
Ignore:
Timestamp:
01/14/03 23:08:02 (6 years ago)
Author:
sholloway
Message:

*** empty log message ***

Files:

Legend:

Unmodified
Added
Removed
Modified
Copied
Moved

  • trunk/RBTelepathy/RBTelepathy/Packet/Elements.py

    r398 r402  
    7777        return ''.join(results) 
    7878 
     79#~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ 
     80 
     81class Error(PacketElementBase): 
     82    #~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ 
     83    #~ OnStreamData adaptor 
     84    #~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ 
     85 
     86    def OnStreamData(self, data, OnPacketComplete=None): 
     87        if OnPacketComplete is not None: 
     88            OnPacketComplete(self) 
     89        return data 
     90 
     91    def GetStreamHeader(self): 
     92        return self._toXML() 
     93 
     94    def GetStreamData(self): 
     95        return '' 
     96 

  • trunk/RBTelepathy/RBTelepathy/Packet/StreamElements.py

    r391 r402  
    5353    _default_attributes = { 
    5454        'format':'length', 
    55         'size':'0', 
     55        #'size': None, 
    5656        } 
    5757 

  • trunk/RBTelepathy/RBTelepathy/Packet/__init__.py

    r398 r402  
    2525 
    2626from RBFoundation.XMLClassBuilder import ElementFactory as EF 
     27from RBFoundation import XMLClassBuilder 
    2728from RBFoundation.XMLObjectify import BaseObjectifiedXML, ObjectifiedXML 
    2829import Builder 
    …… 
    3233#~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ 
    3334#~ Constants / Variables / Etc.  
     35#~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ 
     36 
     37StandardNamespaceSynonyms = { 
     38    'RBMessaging': 'http://namespaces.runeblade.com/RBMessaging', 
     39    None: 'http://namespaces.runeblade.com/RBMessaging', 
     40    } 
     41 
     42#~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ 
     43#~ Definitions  
    3444#~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ 
    3545 
    …… 
    4656        'terminator': StreamElements.TerminatedStream, 
    4757        'term': StreamElements.TerminatedStream, 
     58 
     59        None: StreamElements.LengthStream, 
    4860        } 
    4961 
    …… 
    5365 
    5466    def __call__(self, owner, parent, node, attributes, namespacemap): 
    55         format = attributes['format'] 
     67        format = attributes.get('format') 
    5668        return self.StreamFormatTable[format] 
    5769 
    5870#~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ 
    5971 
    60 StandardElementFactories = { 
    61     ## Long names 
    62     ('http://namespaces.runeblade.com/RBMessaging', 'message'): EF.Static(MessageElements.Message), 
    63     ('http://namespaces.runeblade.com/RBMessaging', 'stream'): StreamFormatFactory(), 
    64     ('http://namespaces.runeblade.com/RBMessaging', 'error'): EF.Static(ObjectifiedXML), 
    65  
    66     ('http://namespaces.runeblade.com/RBMessaging', ): EF.Static(ObjectifiedXML), 
    67     None: EF.Static(ObjectifiedXML), 
    68     } 
     72class MessageErrorFactory(object): 
     73    def __call__(self, owner, parent, node, attributes, namespacemap): 
     74        raise KeyError, '"%s" is not a valid subchild of "%s"' % (node[1], parent.__node__) 
    6975 
    7076#~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ 
    7177 
    72 StandardNamespaceSynonyms = { 
    73     'RBMessaging': 'http://namespaces.runeblade.com/RBMessaging', 
    74     None: 'http://namespaces.runeblade.com/RBMessaging', 
    75     } 
     78class StandardMessage(MessageElements.Message, XMLClassBuilder.XMLClassBuilderBaseMixin): 
     79    ElementFactories = XMLClassBuilder.ElementFactorySet({ 
     80            ('http://namespaces.runeblade.com/RBMessaging', 'stream'): StreamFormatFactory(), 
     81            ('http://namespaces.runeblade.com/RBMessaging', 'to'): EF.Static(ObjectifiedXML), 
     82            ('http://namespaces.runeblade.com/RBMessaging', ): MessageErrorFactory(), 
     83            None: EF.Static(ObjectifiedXML), 
     84        }) 
     85 
     86    NamespaceSynonyms = StandardNamespaceSynonyms 
     87    DefaultNamespace = NamespaceSynonyms[None] 
     88 
     89    def _xmlChildFactory(self, owner, parent, node, attributes, namespacemap): 
     90        return self.ElementFactorySet._GetElementFactory(owner, parent, node, attributes, namespacemap) 
    7691 
    7792#~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ 
    …… 
    8095 
    8196class StandardStreamPacketBuilder(Builder.StreamPacketBuilder): 
    82     ElementFactories = Builder.StreamPacketBuilder.ElementFactories.copy() 
    83     ElementFactories.update(StandardElementFactories) 
     97    ElementFactories = XMLClassBuilder.ElementFactorySet({ 
     98            ('http://namespaces.runeblade.com/RBMessaging', 'message'): EF.Static(StandardMessage), 
     99            ('http://namespaces.runeblade.com/RBMessaging', 'error'): EF.Static(MessageElements.Error), 
     100        }) 
    84101 
    85     NamespaceSynonyms = Builder.StreamPacketBuilder.NamespaceSynonyms.copy() 
    86     NamespaceSynonyms.update(StandardNamespaceSynonyms) 
    87  
     102    NamespaceSynonyms = StandardNamespaceSynonyms 
    88103    DefaultNamespace = NamespaceSynonyms[None] 
    89104 
    90 #~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ 
    91 #~ Testing  
    92 #~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ 
    93  
    94 if __name__=='__main__': 
    95     print "Testing..." 
    96     pb = StandardStreamPacketBuilder() 
    97     info = '''<?xml version='1.0'?> 
    98 <msg> 
    99     <to>account@network/resource?type=mobile</to> 
    100     <stream format='term'/> 
    101     <stream format='len' size='10'/> 
    102 </msg>''' 
    103  
    104     def printit(packet): 
    105         print 
    106         print packet.GetStreamInfo() 
    107         print repr(packet.GetStreamData()) 
    108         print 
    109         print 'Stream Content:' 
    110         for each in packet.datastreams: 
    111             print '   ', repr(each.StreamContent) 
    112         print 
    113  
    114     packet = pb.BuildPacket(info) 
    115  
    116     data = '''Some real text\x04ThereAre10@Trash@''' 
    117     packet.OnStreamData(data) 
    118  
    119     printit(packet) 
    120  
    121     newdata = ['New terminated data', 'New length-based data'] 
    122     for stream, nd in zip(packet.datastreams, newdata): 
    123         stream.StreamContent = nd 
    124  
    125     printit(packet) 
    126  
    127     print "Test complete." 

  • trunk/RBTelepathy/RBTelepathy/Stream/Protocol.py

    r401 r402  
    2626from RBMessaging.Packet import StandardStreamPacketBuilder 
    2727from xml.sax.saxutils import escape as xmlescape 
     28from xml.sax.saxutils import quoteattr as xmlquoteattr  
     29import logging 
    2830 
    2931#~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ 
    …… 
    3941    data = None 
    4042    currentpacket = None 
     43    packetbuilder = None 
     44    stream = None 
     45    log = logging.getLogger('.'.join((__name__, 'StreamProtocol'))) 
    4146 
    4247    #~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ 
    …… 
    4449    #~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ 
    4550 
     51    def __init__(self, PacketBuilderClass=StandardStreamPacketBuilder): 
     52        self.packetbuilder = PacketBuilderClass() 
     53        self.log.debug('StreamProtocol created') 
     54 
     55    def __del__(self): 
     56        self.log.debug('StreamProtocol destroyed') 
     57 
    4658    def SendPacket(self, packet): 
    47         data = ''.join((packet.GetStreamHeader(), self.delimiter, packet.GetStreamData())) 
    48         self.stream.write(data) 
     59        if isinstance(packet, (str, unicode)): 
     60            streamdata = packet + self.delimiter 
     61        else: 
     62            streamdata = ''.join((packet.GetStreamHeader(), self.delimiter, packet.GetStreamData())) 
     63        self.log.debug('Sent %d bytes', len(streamdata)) 
     64        self.stream.write(streamdata) 
    4965 
    5066    def OnRecvStreamData(self, streamdata): 
    51         if self.data: self.data += streamdata 
    52         else: self.data = streamdata 
     67        self.log.debug('Received %d bytes', len(streamdata)) 
     68        if self.data:  
     69            self.data += streamdata 
     70        else:  
     71            self.data = streamdata 
    5372 
    54         if self.currentpacket is None: 
    55             try:  
    56                 # Try to seperate the information packet from the data packet 
    57                 info, self.data = self.data.split(self.delimiter, 1) 
    58             except ValueError:  
    59                 pass # We don't have a complete info packet yet... 
     73        while self.data: 
     74            # While there is still data to be processed 
     75            if self.currentpacket is None: 
     76                try:  
     77                    # Try to seperate the information packet from the data packet 
     78                    info, self.data = self.data.split(self.delimiter, 1) 
     79                except ValueError:  
     80                    # We don't have a complete info packet yet.  
     81                    # Break out of the while self.data loop 
     82                    break  
     83                else: 
     84                    # We now have our information packet in string form 
     85                    try: 
     86                        # Build the packet from the string 
     87                        self.currentpacket = self.packetbuilder.BuildPacket(info) 
     88                        if self.currentpacket is None: 
     89                            # We expected a packet, so not getting one is an error 
     90                            self.StreamError('Incomplete XML header') 
     91                        else: 
     92                            # Good... send the rest of the data as potential stream data 
     93                            self.data = self.currentpacket.OnStreamData(self.data, self._OnPacketComplete) 
     94                    except Exception, e: 
     95                        # Something happened, report to the client 
     96                        self.StreamError('%s: %s' % (e.__class__.__name__, str(e))) 
     97                        self.log.error(e) 
    6098            else: 
    61                 # We now have our information packet... we can figure out what we 
    62                 # are supposed to do with the content stream now 
    63                 try: 
    64                     self.currentpacket = self._BuildPacket(info) 
    65                     if self.currentpacket is None: 
    66                         self.OnStreamError('Incomplete XML Header:\n' + info) 
    67                     else: 
    68                         self.data = self.currentpacket.OnStreamData(self.data, self._OnPacketComplete) 
    69                 except Exception, e: 
    70                     self.OnStreamError(e) 
    71         else: 
    72             self.data = self.currentpacket.OnStreamData(self.data, self._OnPacketComplete) 
     99                self.data = self.currentpacket.OnStreamData(self.data, self._OnPacketComplete) 
     100 
     101 
     102    def StreamError(self, error='Unknown Error', errortype='stream', namespace=None): 
     103        if namespace is None: 
     104            try: 
     105                namespace = self.packetbuilder.DefaultNamespace 
     106            except AttributeError: 
     107                namespace = StandardStreamPacketBuilder.DefaultNamespace 
     108        namespace = 'xmlns=%s' % (xmlquoteattr(namespace),) 
     109 
     110        if errortype is not None: 
     111            errortype = 'type=%s' % (xmlquoteattr(errortype),) 
     112        else: errortype = '' 
     113 
     114        error = xmlescape(error) 
     115 
     116        self.log.info('Stream error from client connection.  Shutting down.') 
     117        self.stream.write('''<error %s %s>%s</error>%s''' % (namespace, errortype, error, self.delimiter)) 
     118        self.stream.shutdown() 
     119 
     120    def OnPacketComplete(self, packet): 
     121        pass 
     122 
     123    #~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ 
     124    #~ Protected Methods  
     125    #~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ 
    73126 
    74127    def _OnPacketComplete(self, packet): 
    …… 
    76129        self.OnPacketComplete(packet) 
    77130 
    78     def OnPacketComplete(self, packet): 
    79         pass 
    80  
    81     def OnStreamError(self, error=None): 
    82         self.stream.write('''<error xmlns='%s'>%s</error>%s''' % (StandardStreamPacketBuilder.DefaultNamespace, xmlescape(str(error or 'Unknown')), self.delimiter)) 
    83         self.stream.shutdown() 
    84  
    85     #~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ 
    86     #~ Protected Methods  
    87     #~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ 
    88  
    89     def _BuildPacket(self, info): 
    90         try: 
    91             packetbuilder = self.packetbuilder 
    92         except AttributeError: 
    93             packetbuilder = self.packetbuilder = StandardStreamPacketBuilder() 
    94         return self.packetbuilder.BuildPacket(info) 
    95