src/python/osrf/net.py

   1 # -----------------------------------------------------------------------
   2 # Copyright (C) 2007  Georgia Public Library Service
   3 # Bill Erickson <billserickson@gmail.com>
   4 #
   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.
   9 #
  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 # -----------------------------------------------------------------------
  15
  16
  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 import osrf.log
  22 import os, time, threading
  23
  24 THREAD_SESSIONS = {}
  25
  26 # - log jabber activity (for future reference)
  27 #import logging
  28 #logger=logging.getLogger()
  29 #logger.addHandler(logging.StreamHandler())
  30 #logger.addHandler(logging.FileHandler('j.log'))
  31 #logger.setLevel(logging.DEBUG)
  32
  33 def set_network_handle(handle):
  34     """ Sets the thread-specific network handle"""
  35     THREAD_SESSIONS[threading.currentThread().getName()] = handle
  36
  37 def get_network_handle():
  38     """ Returns the thread-specific network connection handle."""
  39     return THREAD_SESSIONS.get(threading.currentThread().getName())
  40
  41
  42 class NetworkMessage(object):
  43     """Network message
  44
  45     attributes:
  46
  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
  52     """
  53
  54     def __init__(self, message=None, **args):
  55         if message:
  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')
  62             else:
  63                 self.sender = message.get_from().as_utf8()
  64             self.locale = None # XXX fix me good
  65         else:
  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']
  76
  77 class Network(JabberClient):
  78     def __init__(self, **args):
  79         self.isconnected = False
  80
  81         # Create a unique jabber resource
  82         resource = 'python'
  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)
  88
  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']))
  92
  93         #initialize the superclass
  94         JabberClient.__init__(self, self.jid, args['password'], args['host'])
  95         self.queue = []
  96
  97         self.receive_callback = None
  98
  99     def connect(self):
 100         JabberClient.connect(self)
 101         while not self.isconnected:
 102             stream = self.get_stream()
 103             act = stream.loop_iter(10)
 104             if not act:
 105                 self.idle()
 106
 107     def set_receive_callback(self, func):
 108         """The callback provided is called when a message is received.
 109
 110             The only argument to the function is the received message. """
 111         self.receive_callback = func
 112
 113     def session_started(self):
 114         osrf.log.osrfLogInfo("Successfully connected to the opensrf network")
 115         self.authenticated()
 116         self.stream.set_message_handler("normal", self.message_received)
 117         self.isconnected = True
 118
 119     def send(self, message):
 120         """Sends the provided network message."""
 121         osrf.log.osrfLogInternal("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)
 126
 127     def message_received(self, stanza):
 128         """Handler for received messages."""
 129         if stanza.get_type()=="headline":
 130             return True
 131         # check for errors
 132         osrf.log.osrfLogInternal("jabber received message from %s : %s"
 133             % (stanza.get_from().as_utf8(), stanza.get_body()))
 134         self.queue.append(NetworkMessage(stanza))
 135         return True
 136
 137     def recv(self, timeout=120):
 138         """Attempts to receive a message from the network.
 139
 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
 143         returned.
 144         """
 145
 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
 151                 timeout -= endtime
 152                 osrf.log.osrfLogInternal("exiting stream loop after %s seconds. "
 153                     "act=%s, queue size=%d" % (str(endtime), act, len(self.queue)))
 154                 if not act:
 155                     self.idle()
 156
 157         # if we've acquired a message, handle it
 158         msg = None
 159         if len(self.queue) > 0:
 160             msg = self.queue.pop(0)
 161             if self.receive_callback:
 162                 self.receive_callback(msg)
 163
 164         return msg
 165
 166
 167