Mercurial > libervia-backend

comparison sat.tac @ 13:bd9e9997d540

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
wokkel integration (not finished yet)
author Goffi <goffi@goffi.org>
date Fri, 30 Oct 2009 17:38:27 +0100
parents ef8060d365cb
children a62d7d453f22
comparison
equal deleted inserted replaced
12:ef8060d365cb 13:bd9e9997d540
22 22
23 from twisted.application import internet, service 23 from twisted.application import internet, service
24 from twisted.internet import glib2reactor, protocol, task 24 from twisted.internet import glib2reactor, protocol, task
25 glib2reactor.install() 25 glib2reactor.install()
26 26
27 from twisted.words.protocols.jabber import client, jid, xmlstream, error 27 from twisted.words.protocols.jabber import jid, xmlstream, error
28 from twisted.words.xish import domish 28 from twisted.words.xish import domish
29 29
30 from twisted.internet import reactor 30 from twisted.internet import reactor
31 import pdb 31 import pdb
32
33 from wokkel import client, disco, xmppim
32 34
33 from sat_bridge.DBus import DBusBridge 35 from sat_bridge.DBus import DBusBridge
34 import logging 36 import logging
35 from logging import debug, info, error 37 from logging import debug, info, error
36 38
45 logging.basicConfig(level=logging.DEBUG, 47 logging.basicConfig(level=logging.DEBUG,
46 format='%(message)s') 48 format='%(message)s')
47 ### 49 ###
48 50
49 51
50 52 class SatXMPPClient(client.XMPPClient):
51 53
52 class SAT: 54 def __init__(self, jid, password, host=None, port=5222):
55 client.XMPPClient.__init__(self, jid, password, host, port)
56 self.factory.clientConnectionLost = self.connectionLost
57 self.__connected=False
58
59 def _authd(self, xmlstream):
60 print "SatXMPPClient"
61 client.XMPPClient._authd(self, xmlstream)
62 self.__connected=True
63 print "********** CONNECTED **********"
64 self.streamInitialized()
65
66 def streamInitialized(self):
67 """Called after _authd"""
68 self.keep_alife = task.LoopingCall(self.xmlstream.send, " ") #Needed to avoid disconnection (specially with openfire)
69 self.keep_alife.start(180)
70
71 def isConnected(self):
72 return self.__connected
73
74 def connectionLost(self, connector, unused_reason):
75 print "********** DISCONNECTED **********"
76 try:
77 self.keep_alife.stop()
78 except AttributeError:
79 debug("No keep_alife")
80
81
82 class SatMessageProtocol(xmppim.MessageProtocol):
83
84 def __init__(self, host):
85 xmppim.MessageProtocol.__init__(self)
86 self.host = host
87
88 def onMessage(self, message):
89 debug (u"got_message from: %s", message["from"])
90 for e in message.elements():
91 if e.name == "body":
92 self.host.bridge.newMessage(message["from"], e.children[0])
93 self.host.memory.addToHistory(self.host.me, jid.JID(message["from"]), self.host.me, "chat", e.children[0])
94 break
95
96 class SatRosterProtocol(xmppim.RosterClientProtocol):
97
98 def __init__(self, host):
99 xmppim.RosterClientProtocol.__init__(self)
100 self.host = host
101
102 def rosterCb(self, roster):
103 for jid, item in roster.iteritems():
104 info ("new contact in roster list: %s", jid)
105 #FIXME: fill attributes
106 self.host.memory.addContact(jid, {}, item.groups)
107 self.host.bridge.newContact(jid, {}, item.groups)
108
109 def requestRoster(self):
110 """ ask the server for Roster list """
111 debug("requestRoster")
112 self.getRoster().addCallback(self.rosterCb)
113
114 def removeItem(self, to):
115 """Remove a contact from roster list"""
116 to_jid=jid.JID(to)
117 xmppim.RosterClientProtocol.removeItem(self, to_jid)
118 #TODO: check IQ result
119
120 def addItem(self, to):
121 """Add a contact to roster list"""
122 to_jid=jid.JID(to)
123 xmppim.RosterClientProtocol.addItem(self, to_jid)
124 #TODO: check IQ result
125
126 class SatPresenceProtocol(xmppim.PresenceClientProtocol):
127
128 def __init__(self, host):
129 xmppim.PresenceClientProtocol.__init__(self)
130 self.host = host
131
132 def availableReceived(self, entity, show=None, statuses=None, priority=0):
133 info ("presence update for [%s]", entity)
134
135 ### we check if the status is not about subscription ###
136 #FIXME: type is not needed anymore
137 #TODO: management of differents statuses (differents languages)
138 status = statuses.values()[0] if len(statuses) else ""
139 self.host.memory.addPresenceStatus(entity.full(), "", show or "",
140 status or "", int(priority))
141
142 #now it's time to notify frontends
143 pdb.set_trace()
144 self.host.bridge.presenceUpdate(entity.full(), "", show or "",
145 status or "", int(priority))
146
147 def unavailableReceived(self, entity, statuses=None):
148 #TODO: management of differents statuses (differents languages)
149 status = statuses.values()[0] if len(statuses) else ""
150 self.host.memory.addPresenceStatus(entity.full(), "unavailable", "",
151 status or "", 0)
152
153 #now it's time to notify frontends
154 self.host.bridge.presenceUpdate(entity.full(), "unavailable", "",
155 status or "", 0)
156
157
158 def subscribedReceived(self, entity):
159 debug ("subscription approved for [%s]" % entity)
160
161 def unsubscribedReceived(self, entity):
162 debug ("unsubscription confirmed for [%s]" % entity)
163
164 def subscribeReceived(self, entity):
165 #FIXME: auto answer for subscribe request, must be checked !
166 debug ("subscription request for [%s]" % entity)
167 self.subscribed(entity)
168
169 def unsubscribeReceived(self, entity):
170 debug ("unsubscription asked for [%s]" % entity)
171
172 class SAT(service.Service):
53 173
54 def __init__(self): 174 def __init__(self):
55 #self.reactor=reactor 175 #self.reactor=reactor
56 self.memory=Memory() 176 self.memory=Memory()
57 self.server_features=[] #XXX: temp dic, need to be transfered into self.memory in the future 177 self.server_features=[] #XXX: temp dic, need to be transfered into self.memory in the future
58 self.connected=False #FIXME: use twisted var instead
59 178
60 self._iq_cb_map = {} #callback called when ns is found on IQ 179 self._iq_cb_map = {} #callback called when ns is found on IQ
61 self._waiting_conf = {} #callback called when a confirmation is received 180 self._waiting_conf = {} #callback called when a confirmation is received
62 self._progress_cb_map = {} #callback called when a progress is requested (key = progress id) 181 self._progress_cb_map = {} #callback called when a progress is requested (key = progress id)
63 self.plugins = {} 182 self.plugins = {}
96 plug_info = mod.PLUGIN_INFO 215 plug_info = mod.PLUGIN_INFO
97 info ("importing plugin: %s", plug_info['name']) 216 info ("importing plugin: %s", plug_info['name'])
98 self.plugins[plug_info['import_name']] = getattr(mod, plug_info['main'])(self) 217 self.plugins[plug_info['import_name']] = getattr(mod, plug_info['main'])(self)
99 218
100 def connect(self): 219 def connect(self):
101 if (self.connected): 220 if (self.isConnected()):
102 info("already connected !") 221 info("already connected !")
103 return 222 return
104 print "connecting..." 223 print "connecting..."
105 reactor.connectTCP(self.memory.getParamV("Server", "Connection"), 5222, self.factory) 224 self.me = jid.JID(self.memory.getParamV("JabberID", "Connection"))
225 self.xmppclient = SatXMPPClient(self.me, self.memory.getParamV("Password", "Connection"),
226 self.memory.getParamV("Server", "Connection"), 5222)
227 self.xmppclient.streamInitialized = self.streamInitialized
228
229 self.messageProt = SatMessageProtocol(self)
230 self.messageProt.setHandlerParent(self.xmppclient)
231
232 self.roster = SatRosterProtocol(self)
233 self.roster.setHandlerParent(self.xmppclient)
234
235 self.presence = SatPresenceProtocol(self)
236 self.presence.setHandlerParent(self.xmppclient)
237
238 self.xmppclient.startService()
106 239
107 def disconnect(self): 240 def disconnect(self):
108 if (not self.connected): 241 if (not self.isConnected()):
109 info("not connected !") 242 info("not connected !")
110 return 243 return
111 info("Disconnecting...") 244 info("Disconnecting...")
112 self.factory.stopTrying() 245 self.xmppclient.stopService()
113 if self.xmlstream: 246
114 self.xmlstream.sendFooter() 247 def startService(self):
115 248 info("Salut à toi ô mon frère !")
116 def getService(self): 249 self.connect()
117 print "GetService !" 250
118 """if (self.connected): 251 def stopService(self):
119 info("already connected !") 252 info("Salut aussi à Rantanplan")
120 return"""
121 info("Getting service...")
122 self.me = jid.JID(self.memory.getParamV("JabberID", "Connection"))
123 self.factory = client.XMPPClientFactory(self.me, self.memory.getParamV("Password", "Connection"))
124 self.factory.clientConnectionLost = self.connectionLost
125 self.factory.addBootstrap(xmlstream.STREAM_AUTHD_EVENT,self.authd)
126 self.factory.addBootstrap(xmlstream.INIT_FAILED_EVENT,self.failed)
127 return internet.TCPClient(self.memory.getParamV("Server", "Connection"), 5222, self.factory)
128 253
129 def run(self): 254 def run(self):
130 debug("running app") 255 debug("running app")
131 reactor.run() 256 reactor.run()
132 257
133 def stop(self): 258 def stop(self):
134 debug("stopping app") 259 debug("stopping app")
135 reactor.stop() 260 reactor.stop()
136 261
137 def authd(self,xmlstream): 262 def streamInitialized(self):
138 self.xmlstream=xmlstream 263 """Called when xmlstream is OK"""
139 roster=client.IQ(xmlstream,'get') 264 SatXMPPClient.streamInitialized(self.xmppclient)
140 roster.addElement(('jabber:iq:roster', 'query')) 265 debug ("XML stream is initialized")
141 roster.addCallback(self.rosterCb) 266 self.xmlstream = self.xmppclient.xmlstream
142 roster.send() 267 self.me = self.xmppclient.jid #in case of the ressource has changed
143 debug("server = %s",self.memory.getParamV("Server", "Connection")) 268
144 269 self.roster.requestRoster()
270
271 self.presence.available()
272
273 #FIXME:tmp
274 self.xmlstream.addObserver("/iq[@type='set' or @type='get']", self.iqCb)
275 """
145 ###FIXME: tmp disco ### 276 ###FIXME: tmp disco ###
277 #self.discoHandler = disco.discoHandler()
146 self.memory.registerFeature("http://jabber.org/protocol/disco#info") 278 self.memory.registerFeature("http://jabber.org/protocol/disco#info")
147 self.disco(self.memory.getParamV("Server", "Connection"), self.serverDisco) 279 self.disco(self.memory.getParamV("Server", "Connection"), self.serverDisco)
148
149
150 #we now send our presence status 280 #we now send our presence status
151 self.setPresence(status="Online")
152 281
153 # add a callback for the messages 282 # add a callback for the messages
154 xmlstream.addObserver('/message', self.gotMessage)
155 xmlstream.addObserver('/presence', self.presenceCb)
156 xmlstream.addObserver("/iq[@type='set' or @type='get']", self.iqCb)
157 print "********** CONNECTED **********"
158 self.connected=True
159 self.keep_alife = task.LoopingCall(self.xmlstream.send, " ") #Needed to avoid disconnection (specially with openfire)
160 self.keep_alife.start(180)
161 283
162 #reactor.callLater(2,self.sendFile,"goffi2@jabber.goffi.int/Psi", "/tmp/fakefile") 284 #reactor.callLater(2,self.sendFile,"goffi2@jabber.goffi.int/Psi", "/tmp/fakefile")
163 285 """
164 def connectionLost(self, connector, unused_reason):
165 print "********** DISCONNECTED **********"
166 if self.keep_alife:
167 self.keep_alife.stop()
168 self.connected=False
169 286
170 287
171 def sendMessage(self,to,msg,type='chat'): 288 def sendMessage(self,to,msg,type='chat'):
172 #FIXME: check validity of recipient 289 #FIXME: check validity of recipient
173 debug("Sending jabber message to %s...", to) 290 debug("Sending jabber message to %s...", to)
184 """set wanted paramater and notice observers""" 301 """set wanted paramater and notice observers"""
185 info ("setting param: %s=%s in namespace %s", name, value, namespace) 302 info ("setting param: %s=%s in namespace %s", name, value, namespace)
186 self.memory.setParam(name, value, namespace) 303 self.memory.setParam(name, value, namespace)
187 self.bridge.paramUpdate(name, value, namespace) 304 self.bridge.paramUpdate(name, value, namespace)
188 305
189 def setRoster(self, to):
190 """Add a contact to roster list"""
191 to_jid=jid.JID(to)
192 roster=client.IQ(self.xmlstream,'set')
193 query=roster.addElement(('jabber:iq:roster', 'query'))
194 item=query.addElement("item")
195 item.attributes["jid"]=to_jid.userhost()
196 roster.send()
197 #TODO: check IQ result
198
199 def delRoster(self, to):
200 """Remove a contact from roster list"""
201 to_jid=jid.JID(to)
202 roster=client.IQ(self.xmlstream,'set')
203 query=roster.addElement(('jabber:iq:roster', 'query'))
204 item=query.addElement("item")
205 item.attributes["jid"]=to_jid.userhost()
206 item.attributes["subscription"]="remove"
207 roster.send()
208 #TODO: check IQ result
209
210
211 def failed(self,xmlstream): 306 def failed(self,xmlstream):
212 debug("failed: %s", xmlstream.getErrorMessage()) 307 debug("failed: %s", xmlstream.getErrorMessage())
213 debug("failed: %s", dir(xmlstream)) 308 debug("failed: %s", dir(xmlstream))
214 309
215 def isConnected(self): 310 def isConnected(self):
216 return self.connected 311 try:
312 if self.xmppclient.isConnected():
313 return True
314 except AttributeError:
315 #xmppclient not available
316 pass
317 return False
217 318
218 ## jabber methods ## 319 ## jabber methods ##
219 320
220 def disco (self, item, callback, node=None): 321 def disco (self, item, callback, node=None):
221 """XEP-0030 Service discovery Feature.""" 322 """XEP-0030 Service discovery Feature."""
222 disco=client.IQ(self.xmlstream,'get') 323 """disco=client.IQ(self.xmlstream,'get')
223 disco["from"]=self.me.full() 324 disco["from"]=self.me.full()
224 disco["to"]=item 325 disco["to"]=item
225 disco.addElement(('http://jabber.org/protocol/disco#info', 'query')) 326 disco.addElement(('http://jabber.org/protocol/disco#info', 'query'))
226 disco.addCallback(callback) 327 disco.addCallback(callback)
227 disco.send() 328 disco.send()"""
228 329
229 330
230 def setPresence(self, to="", type="", show="", status="", priority=0): 331 def setPresence(self, to="", type="", show="", status="", priority=0):
231 """Send our presence information""" 332 """Send our presence information"""
232 presence = domish.Element(('jabber:client', 'presence'))
233 if not type in ["", "unavailable", "subscribed", "subscribe", 333 if not type in ["", "unavailable", "subscribed", "subscribe",
234 "unsubscribe", "unsubscribed", "prob", "error"]: 334 "unsubscribe", "unsubscribed", "prob", "error"]:
235 error("Type error !") 335 error("Type error !")
236 #TODO: throw an error 336 #TODO: throw an error
237 return 337 return
238 338 to_jid=jid.JID(to)
239 if to: 339 #TODO: refactor subscription bridge API
240 presence.attributes["to"]=to 340 if type=="":
241 if type: 341 self.presence.available(to_jid, show, status, priority)
242 presence.attributes["type"]=type 342 elif type=="subscribe":
243 343 self.presence.subscribe(to_jid)
244 for element in ["show", "status", "priority"]: 344 elif type=="subscribed":
245 if locals()[element]: 345 self.presence.subscribed(to_jid)
246 presence.addElement(element).addContent(unicode(locals()[element])) 346 elif type=="unsubscribe":
247 347 self.presence.unsubscribe(to_jid)
248 self.xmlstream.send(presence) 348 elif type=="unsubscribed":
349 self.presence.unsubscribed(to_jid)
350
249 351
250 def addContact(self, to): 352 def addContact(self, to):
251 """Add a contact in roster list""" 353 """Add a contact in roster list"""
252 to_jid=jid.JID(to) 354 to_jid=jid.JID(to)
253 self.setRoster(to_jid.userhost()) 355 self.roster.addItem(to_jid.userhost())
254 self.setPresence(to_jid.userhost(), "subscribe") 356 self.setPresence(to_jid.userhost(), "subscribe")
255 357
256 def delContact(self, to): 358 def delContact(self, to):
257 """Remove contact from roster list""" 359 """Remove contact from roster list"""
258 to_jid=jid.JID(to) 360 to_jid=jid.JID(to)
259 self.delRoster(to_jid.userhost()) 361 self.roster.removeItem(to_jid.userhost())
260 self.bridge.contactDeleted(to) 362 self.bridge.contactDeleted(to)
261 363
262 def gotMessage(self,message):
263 debug (u"got_message from: %s", message["from"])
264 for e in message.elements():
265 if e.name == "body":
266 self.bridge.newMessage(message["from"], e.children[0])
267 self.memory.addToHistory(self.me, jid.JID(message["from"]), self.me, "chat", e.children[0])
268 break
269 364
270 ## callbacks ## 365 ## callbacks ##
271 366
272 def add_IQ_cb(self, ns, cb): 367 def add_IQ_cb(self, ns, cb):
273 """Add an IQ callback on namespace ns""" 368 """Add an IQ callback on namespace ns"""
284 uri = stanza.firstChildElement().uri 379 uri = stanza.firstChildElement().uri
285 if self._iq_cb_map.has_key(uri): 380 if self._iq_cb_map.has_key(uri):
286 self._iq_cb_map[uri](stanza) 381 self._iq_cb_map[uri](stanza)
287 #TODO: manage errors stanza 382 #TODO: manage errors stanza
288 383
289 def presenceCb(self, elem): 384
290 info ("presence update for [%s]", elem.getAttribute("from"))
291 debug("\n\nXML=\n%s\n\n", elem.toXml())
292 presence={}
293 presence["jid"]=elem.getAttribute("from")
294 presence["type"]=elem.getAttribute("type") or ""
295 presence["show"]=""
296 presence["status"]=""
297 presence["priority"]=0
298
299 for item in elem.elements():
300 if presence.has_key(item.name):
301 presence[item.name]=item.children[0]
302
303 ### we check if the status is not about subscription ###
304 #TODO: check that from jid is one we wants to subscribe (ie: check a recent subscription asking)
305 if jid.JID(presence["jid"]).userhost()!=self.me.userhost():
306 if presence["type"]=="subscribed":
307 debug ("subscription answer")
308 elif presence["type"]=="unsubscribed":
309 debug ("unsubscription answer")
310 elif presence["type"]=="subscribe":
311 #FIXME: auto answer for subscribe request, must be checked !
312 debug ("subscription request")
313 self.setPresence(to=presence["jid"], type="subscribed")
314 else:
315 #We keep presence information only if it is not for subscription
316 self.memory.addPresenceStatus(presence["jid"], presence["type"], presence["show"],
317 presence["status"], int(presence["priority"]))
318
319 #now it's time to notify frontends
320 self.bridge.presenceUpdate(presence["jid"], presence["type"], presence["show"],
321 presence["status"], int(presence["priority"]))
322
323 def rosterCb(self,roster):
324 for contact in roster.firstChildElement().elements():
325 info ("new contact in roster list: %s", contact['jid'])
326 #and now the groups
327 groups=[]
328 for group in contact.elements():
329 if group.name!="group":
330 error("Unexpected element !")
331 break
332 groups.append(str(group))
333 self.memory.addContact(contact['jid'], contact.attributes, groups)
334 self.bridge.newContact(contact['jid'], contact.attributes, groups)
335 385
336 def serverDisco(self, disco): 386 def serverDisco(self, disco):
337 """xep-0030 Discovery Protocol.""" 387 """xep-0030 Discovery Protocol."""
338 for element in disco.firstChildElement().elements(): 388 for element in disco.firstChildElement().elements():
339 if element.name=="feature": 389 if element.name=="feature":
388 #debug("Requested progress for unknown id") 438 #debug("Requested progress for unknown id")
389 return data 439 return data
390 440
391 441
392 application = service.Application('SàT') 442 application = service.Application('SàT')
393 sat = SAT() 443 service = SAT()
394 service = sat.getService()
395 service.setServiceParent(application) 444 service.setServiceParent(application)
396 445
397 446
398 #app.memory.save() #FIXME: not the best place 447 #app.memory.save() #FIXME: not the best place
399 #debug("Good Bye") 448 #debug("Good Bye")