Mercurial > libervia-backend

comparison src/plugins/plugin_xep_0047.py @ 587:952322b1d490

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
Remove trailing whitespaces.
author Emmanuel Gil Peyrot <linkmauve@linkmauve.fr>
date Fri, 18 Jan 2013 17:55:34 +0100
parents 9902ec2d8d9b
children beaf6bec2fcd
comparison
equal deleted inserted replaced
586:6a718ede8be1 587:952322b1d490
70 def profileConnected(self, profile): 70 def profileConnected(self, profile):
71 client = self.host.getClient(profile) 71 client = self.host.getClient(profile)
72 if not client: 72 if not client:
73 raise ProfileNotInCacheError 73 raise ProfileNotInCacheError
74 client.xep_0047_current_stream = {} #key: stream_id, value: data(dict) 74 client.xep_0047_current_stream = {} #key: stream_id, value: data(dict)
75 75
76 def _timeOut(self, sid, profile): 76 def _timeOut(self, sid, profile):
77 """Delecte current_stream id, called after timeout 77 """Delecte current_stream id, called after timeout
78 @param id: id of client.xep_0047_current_stream""" 78 @param id: id of client.xep_0047_current_stream"""
79 info(_("In-Band Bytestream: TimeOut reached for id %s [%s]") % (sid, profile)) 79 info(_("In-Band Bytestream: TimeOut reached for id %s [%s]") % (sid, profile))
80 self._killId(sid, False, "TIMEOUT", profile) 80 self._killId(sid, False, "TIMEOUT", profile)
81 81
82 def _killId(self, sid, success=False, failure_reason="UNKNOWN", profile=None): 82 def _killId(self, sid, success=False, failure_reason="UNKNOWN", profile=None):
83 """Delete an current_stream id, clean up associated observers 83 """Delete an current_stream id, clean up associated observers
84 @param sid: id of client.xep_0047_current_stream""" 84 @param sid: id of client.xep_0047_current_stream"""
85 assert(profile) 85 assert(profile)
86 client = self.host.getClient(profile) 86 client = self.host.getClient(profile)
94 client.xmlstream.removeObserver(client.xep_0047_current_stream[sid]["event_data"], client.xep_0047_current_stream[sid]["observer_cb"]) 94 client.xmlstream.removeObserver(client.xep_0047_current_stream[sid]["event_data"], client.xep_0047_current_stream[sid]["observer_cb"])
95 if client.xep_0047_current_stream[sid]['timer'].active(): 95 if client.xep_0047_current_stream[sid]['timer'].active():
96 client.xep_0047_current_stream[sid]['timer'].cancel() 96 client.xep_0047_current_stream[sid]['timer'].cancel()
97 if client.xep_0047_current_stream[sid].has_key("size"): 97 if client.xep_0047_current_stream[sid].has_key("size"):
98 self.host.removeProgressCB(sid, profile) 98 self.host.removeProgressCB(sid, profile)
99 99
100 file_obj = client.xep_0047_current_stream[sid]['file_obj'] 100 file_obj = client.xep_0047_current_stream[sid]['file_obj']
101 success_cb = client.xep_0047_current_stream[sid]['success_cb'] 101 success_cb = client.xep_0047_current_stream[sid]['success_cb']
102 failure_cb = client.xep_0047_current_stream[sid]['failure_cb'] 102 failure_cb = client.xep_0047_current_stream[sid]['failure_cb']
103 103
104 del client.xep_0047_current_stream[sid] 104 del client.xep_0047_current_stream[sid]
105 105
106 if success: 106 if success:
107 success_cb(sid, file_obj, NS_IBB, profile) 107 success_cb(sid, file_obj, NS_IBB, profile)
108 else: 108 else:
109 failure_cb(sid, file_obj, NS_IBB, failure_reason, profile) 109 failure_cb(sid, file_obj, NS_IBB, failure_reason, profile)
110 110
111 def getProgress(self, sid, data, profile): 111 def getProgress(self, sid, data, profile):
112 """Fill data with position of current transfer""" 112 """Fill data with position of current transfer"""
113 client = self.host.getClient(profile) 113 client = self.host.getClient(profile)
114 if not client: 114 if not client:
115 raise ProfileNotInCacheError 115 raise ProfileNotInCacheError
117 file_obj = client.xep_0047_current_stream[sid]["file_obj"] 117 file_obj = client.xep_0047_current_stream[sid]["file_obj"]
118 data["position"] = str(file_obj.tell()) 118 data["position"] = str(file_obj.tell())
119 data["size"] = str(client.xep_0047_current_stream[sid]["size"]) 119 data["size"] = str(client.xep_0047_current_stream[sid]["size"])
120 except: 120 except:
121 pass 121 pass
122 122
123 def prepareToReceive(self, from_jid, sid, file_obj, size, success_cb, failure_cb, profile): 123 def prepareToReceive(self, from_jid, sid, file_obj, size, success_cb, failure_cb, profile):
124 """Called when a bytestream is imminent 124 """Called when a bytestream is imminent
125 @param from_jid: jid of the sender 125 @param from_jid: jid of the sender
126 @param sid: Stream id 126 @param sid: Stream id
127 @param file_obj: File object where data will be written 127 @param file_obj: File object where data will be written
172 #we reset the timeout: 172 #we reset the timeout:
173 client.xep_0047_current_stream[sid]["timer"].reset(TIMEOUT) 173 client.xep_0047_current_stream[sid]["timer"].reset(TIMEOUT)
174 174
175 #we save the xmlstream, events and observer data to allow observer removal 175 #we save the xmlstream, events and observer data to allow observer removal
176 client.xep_0047_current_stream[sid]["event_data"] = event_data = (IBB_MESSAGE_DATA if stanza=='message' else IBB_IQ_DATA) % sid 176 client.xep_0047_current_stream[sid]["event_data"] = event_data = (IBB_MESSAGE_DATA if stanza=='message' else IBB_IQ_DATA) % sid
177 client.xep_0047_current_stream[sid]["observer_cb"] = observer_cb = self.messageData if stanza=='message' else self.iqData 177 client.xep_0047_current_stream[sid]["observer_cb"] = observer_cb = self.messageData if stanza=='message' else self.iqData
178 event_close = IBB_CLOSE % sid 178 event_close = IBB_CLOSE % sid
179 #we now set the stream observer to look after data packet 179 #we now set the stream observer to look after data packet
180 client.xmlstream.addObserver(event_data, observer_cb, profile = profile) 180 client.xmlstream.addObserver(event_data, observer_cb, profile = profile)
181 client.xmlstream.addOnetimeObserver(event_close, self.streamClosing, profile = profile) 181 client.xmlstream.addOnetimeObserver(event_close, self.streamClosing, profile = profile)
182 #finally, we send the accept stanza 182 #finally, we send the accept stanza
205 IQ.handled=True 205 IQ.handled=True
206 client = self.host.getClient(profile) 206 client = self.host.getClient(profile)
207 if not client: 207 if not client:
208 raise ProfileNotInCacheError 208 raise ProfileNotInCacheError
209 data_elt = IQ.firstChildElement() 209 data_elt = IQ.firstChildElement()
210 210
211 if self._manageDataElt(data_elt, 'iq', IQ['id'], jid.JID(IQ['from']), profile): 211 if self._manageDataElt(data_elt, 'iq', IQ['id'], jid.JID(IQ['from']), profile):
212 #and send a success answer 212 #and send a success answer
213 result = domish.Element((None, 'iq')) 213 result = domish.Element((None, 'iq'))
214 result['type'] = 'result' 214 result['type'] = 'result'
215 result['id'] = IQ['id'] 215 result['id'] = IQ['id']
216 result['to'] = IQ['from'] 216 result['to'] = IQ['from']
217 217
218 client.xmlstream.send(result) 218 client.xmlstream.send(result)
219 219
220 def messageData(self, message_elt, profile): 220 def messageData(self, message_elt, profile):
221 data_elt = message_elt.firstChildElement() 221 data_elt = message_elt.firstChildElement()
222 sid = message_elt.getAttribute('id','') 222 sid = message_elt.getAttribute('id','')
223 self._manageDataElt(message_elt, 'message', sid, jid.JID(message_elt['from']), profile) 223 self._manageDataElt(message_elt, 'message', sid, jid.JID(message_elt['from']), profile)
224 224
270 @param to_jid: addressee 270 @param to_jid: addressee
271 @param xmlstream: XML stream to use to send the error""" 271 @param xmlstream: XML stream to use to send the error"""
272 result = domish.Element((None, 'iq')) 272 result = domish.Element((None, 'iq'))
273 result['type'] = 'result' 273 result['type'] = 'result'
274 result['id'] = iq_id 274 result['id'] = iq_id
275 result['to'] = to_jid 275 result['to'] = to_jid
276 error_el = result.addElement('error') 276 error_el = result.addElement('error')
277 error_el['type'] = 'cancel' 277 error_el['type'] = 'cancel'
278 error_el.addElement(('urn:ietf:params:xml:ns:xmpp-stanzas','not-acceptable')) 278 error_el.addElement(('urn:ietf:params:xml:ns:xmpp-stanzas','not-acceptable'))
279 xmlstream.send(result) 279 xmlstream.send(result)
280 280
321 data = client.xep_0047_current_stream[sid] 321 data = client.xep_0047_current_stream[sid]
322 if iq_elt["type"] == "error": 322 if iq_elt["type"] == "error":
323 warning(_("Transfer failed")) 323 warning(_("Transfer failed"))
324 self.terminateStream(sid, "IQ_ERROR") 324 self.terminateStream(sid, "IQ_ERROR")
325 return 325 return
326 326
327 if data['timer'].active(): 327 if data['timer'].active():
328 data['timer'].cancel() 328 data['timer'].cancel()
329 329
330 buffer = data["file_obj"].read(data["block_size"]) 330 buffer = data["file_obj"].read(data["block_size"])
331 if buffer: 331 if buffer:
332 next_iq_elt = jabber_client.IQ(client.xmlstream,'set') 332 next_iq_elt = jabber_client.IQ(client.xmlstream,'set')
333 next_iq_elt['to'] = data["to"].full() 333 next_iq_elt['to'] = data["to"].full()
334 data_elt = next_iq_elt.addElement('data', NS_IBB) 334 data_elt = next_iq_elt.addElement('data', NS_IBB)
364 else: 364 else:
365 self._killId(sid, True, profile=profile) 365 self._killId(sid, True, profile=profile)
366 366
367 class XEP_0047_handler(XMPPHandler): 367 class XEP_0047_handler(XMPPHandler):
368 implements(iwokkel.IDisco) 368 implements(iwokkel.IDisco)
369 369
370 def __init__(self,parent): 370 def __init__(self,parent):
371 self.plugin_parent = parent 371 self.plugin_parent = parent
372 372
373 def connectionInitialized(self): 373 def connectionInitialized(self):
374 self.xmlstream.addObserver(IBB_OPEN, self.plugin_parent.streamOpening, profile = self.parent.profile) 374 self.xmlstream.addObserver(IBB_OPEN, self.plugin_parent.streamOpening, profile = self.parent.profile)
375 375
376 def getDiscoInfo(self, requestor, target, nodeIdentifier=''): 376 def getDiscoInfo(self, requestor, target, nodeIdentifier=''):
377 return [disco.DiscoFeature(NS_IBB)] 377 return [disco.DiscoFeature(NS_IBB)]