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
22 import os, time, threading
26 # - log jabber activity (for future reference)
28 #logger=logging.getLogger()
29 #logger.addHandler(logging.StreamHandler())
30 #logger.addHandler(logging.FileHandler('j.log'))
31 #logger.setLevel(logging.DEBUG)
33 def set_network_handle(handle):
34 """ Sets the thread-specific network handle"""
35 THREAD_SESSIONS[threading.currentThread().getName()] = handle
37 def get_network_handle():
38 """ Returns the thread-specific network connection handle."""
39 return THREAD_SESSIONS.get(threading.currentThread().getName())
42 class NetworkMessage(object):
47 sender - message sender
48 recipient - message recipient
49 body - the body of the message
50 thread - the message thread
51 locale - locale of the message
54 def __init__(self, message=None, **args):
56 self.body = message.get_body()
57 self.thread = message.get_thread()
58 self.recipient = message.get_to()
59 if message.xmlnode.hasProp('router_from') and \
60 message.xmlnode.prop('router_from') != '':
61 self.sender = message.xmlnode.prop('router_from')
63 self.sender = message.get_from().as_utf8()
64 self.locale = None # XXX fix me good
66 if args.has_key('sender'):
67 self.sender = args['sender']
68 if args.has_key('recipient'):
69 self.recipient = args['recipient']
70 if args.has_key('body'):
71 self.body = args['body']
72 if args.has_key('thread'):
73 self.thread = args['thread']
74 if args.has_key('locale'):
75 self.thread = args['locale']
77 class Network(JabberClient):
78 def __init__(self, **args):
79 self.isconnected = False
81 # Create a unique jabber resource
83 if args.has_key('resource'):
84 resource = args['resource']
85 resource += '_' + gethostname() + ':' + str(os.getpid()) + '_' + \
86 threading.currentThread().getName().lower()
87 self.jid = JID(args['username'], args['host'], resource)
89 osrf.log.logDebug("initializing network with JID %s and host=%s, "
90 "port=%s, username=%s" % (self.jid.as_utf8(), args['host'], \
91 args['port'], args['username']))
93 #initialize the superclass
94 JabberClient.__init__(self, self.jid, args['password'], args['host'])
97 self.receive_callback = None
100 JabberClient.connect(self)
101 while not self.isconnected:
102 stream = self.get_stream()
103 act = stream.loop_iter(10)
107 def set_receive_callback(self, func):
108 """The callback provided is called when a message is received.
110 The only argument to the function is the received message. """
111 self.receive_callback = func
113 def session_started(self):
114 osrf.log.log_info("Successfully connected to the opensrf network")
116 self.stream.set_message_handler("normal", self.message_received)
117 self.isconnected = True
119 def send(self, message):
120 """Sends the provided network message."""
121 osrf.log.log_internal("jabber sending to %s: %s" % \
122 (message.recipient, message.body))
123 msg = Message(None, None, message.recipient, None, None, None, \
124 message.body, message.thread)
125 self.stream.send(msg)
127 def message_received(self, stanza):
128 """Handler for received messages."""
129 if stanza.get_type()=="headline":
132 osrf.log.log_internal("jabber received message from %s : %s"
133 % (stanza.get_from().as_utf8(), stanza.get_body()))
134 self.queue.append(NetworkMessage(stanza))
137 def recv(self, timeout=120):
138 """Attempts to receive a message from the network.
140 timeout - max number of seconds to wait for a message.
141 If a message is received in 'timeout' seconds, the message is passed to
142 the receive_callback is called and True is returned. Otherwise, false is
146 if len(self.queue) == 0:
147 while timeout >= 0 and len(self.queue) == 0:
148 starttime = time.time()
149 act = self.get_stream().loop_iter(timeout)
150 endtime = time.time() - starttime
152 osrf.log.log_internal("exiting stream loop after %s seconds. "
153 "act=%s, queue size=%d" % (str(endtime), act, len(self.queue)))
157 # if we've acquired a message, handle it
159 if len(self.queue) > 0:
160 msg = self.queue.pop(0)
161 if self.receive_callback:
162 self.receive_callback(msg)