1 # -----------------------------------------------------------------------
2 # Copyright (C) 2007 Georgia Public Library Service
3 # Bill Erickson <billserickson@gmail.com>
5 # This program is free software; you can redistribute it and/or
6 # modify it under the terms of the GNU General Public License
7 # as published by the Free Software Foundation; either version 2
8 # of the License, or (at your option) any later version.
10 # This program is distributed in the hope that it will be useful,
11 # but WITHOUT ANY WARRANTY; without even the implied warranty of
12 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
13 # GNU General Public License for more details.
14 # -----------------------------------------------------------------------
17 from pyxmpp.jabber.client import JabberClient
18 from pyxmpp.message import Message
19 from pyxmpp.jid import JID
20 from socket import gethostname
21 from osrf.log import *
25 # - log jabber activity (for future reference)
26 #logger=logging.getLogger()
27 #logger.addHandler(logging.StreamHandler())
28 #logger.addHandler(logging.FileHandler('j.log'))
29 #logger.setLevel(logging.DEBUG)
32 def osrfSetNetworkHandle(handle):
33 """Sets the global network connection handle."""
37 def osrfGetNetworkHandle():
38 """Returns the global network connection handle."""
43 class osrfNetworkMessage(object):
48 sender - message sender
49 to - message recipient
50 body - the body of the message
51 thread - the message thread
54 def __init__(self, message=None, **args):
56 self.body = message.get_body()
57 self.thread = message.get_thread()
58 self.to = message.get_to()
59 if message.xmlnode.hasProp('router_from') and message.xmlnode.prop('router_from') != '':
60 self.sender = message.xmlnode.prop('router_from')
61 else: self.sender = message.get_from().as_utf8()
63 if args.has_key('sender'): self.sender = args['sender']
64 if args.has_key('to'): self.to = args['to']
65 if args.has_key('body'): self.body = args['body']
66 if args.has_key('thread'): self.thread = args['thread']
69 class osrfNetwork(JabberClient):
70 def __init__(self, **args):
71 self.isconnected = False
73 # Create a unique jabber resource
74 resource = 'osrf_client'
75 if args.has_key('resource'):
76 resource = args['resource']
77 resource += '_' + gethostname()+':'+ str(os.getpid())
78 self.jid = JID(args['username'], args['host'], resource)
80 osrfLogDebug("initializing network with JID %s and host=%s, port=%s, username=%s" %
81 (self.jid.as_utf8(), args['host'], args['port'], args['username']))
83 #initialize the superclass
84 JabberClient.__init__(self, self.jid, args['password'], args['host'])
88 JabberClient.connect(self)
89 while not self.isconnected:
90 stream = self.get_stream()
91 act = stream.loop_iter(10)
92 if not act: self.idle()
94 def setRecvCallback(self, func):
95 """The callback provided is called when a message is received.
97 The only argument to the function is the received message. """
98 self.recvCallback = func
100 def session_started(self):
101 osrfLogInfo("Successfully connected to the opensrf network")
103 self.stream.set_message_handler("normal",self.message_received)
104 self.isconnected = True
106 def send(self, message):
107 """Sends the provided network message."""
108 osrfLogInternal("jabber sending to %s: %s" % (message.to, message.body))
109 msg = Message(None, None, message.to, None, None, None, message.body, message.thread)
110 self.stream.send(msg)
112 def message_received(self, stanza):
113 """Handler for received messages."""
114 osrfLogInternal("jabber received a message of type %s" % stanza.get_type())
115 if stanza.get_type()=="headline":
118 osrfLogInternal("jabber received message from %s : %s"
119 % (stanza.get_from().as_utf8(), stanza.get_body()))
120 self.queue.append(osrfNetworkMessage(stanza))
123 def recv(self, timeout=120):
124 """Attempts to receive a message from the network.
126 timeout - max number of seconds to wait for a message.
127 If no message is received in 'timeout' seconds, None is returned. """
130 if len(self.queue) == 0:
131 while timeout >= 0 and len(self.queue) == 0:
132 starttime = time.time()
133 osrfLogInternal("going into stream loop at " + str(starttime))
134 act = self.get_stream().loop_iter(timeout)
135 endtime = time.time() - starttime
137 osrfLogInternal("exiting stream loop after %s seconds" % str(endtime))
138 osrfLogInternal("act = %s : queue length = %d" % (act, len(self.queue)) )
139 if not act: self.idle()
141 # if we've acquired a message, handle it
142 if len(self.queue) > 0:
143 self.recvCallback(self.queue.pop(0))