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 from osrf.log import *
  22 import os, time
  23 import logging
  24
  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)
  30
  31 __network = None
  32 def osrfSetNetworkHandle(handle):
  33         """Sets the global network connection handle."""
  34         global __network
  35         __network = handle
  36
  37 def osrfGetNetworkHandle():
  38         """Returns the global network connection handle."""
  39         global __network
  40         return __network
  41
  42
  43 class osrfNetworkMessage(object):
  44         """Network message
  45
  46         attributes:
  47
  48         sender - message sender
  49         to - message recipient
  50         body - the body of the message
  51         thread - the message thread
  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.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()
  62                 else:
  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']
  67
  68
  69 class osrfNetwork(JabberClient):
  70         def __init__(self, **args):
  71                 self.isconnected = False
  72
  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)
  79
  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']))
  82
  83                 #initialize the superclass
  84                 JabberClient.__init__(self, self.jid, args['password'], args['host'])
  85                 self.queue = []
  86
  87         def connect(self):
  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()
  93
  94         def setRecvCallback(self, func):
  95                 """The callback provided is called when a message is received.
  96
  97                         The only argument to the function is the received message. """
  98                 self.recvCallback = func
  99
 100         def session_started(self):
 101                 osrfLogInfo("Successfully connected to the opensrf network")
 102                 self.authenticated()
 103                 self.stream.set_message_handler("normal",self.message_received)
 104                 self.isconnected = True
 105
 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)
 111
 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":
 116                         return True
 117                 # check for errors
 118                 osrfLogInternal("jabber received message from %s : %s"
 119                         % (stanza.get_from().as_utf8(), stanza.get_body()))
 120                 self.queue.append(osrfNetworkMessage(stanza))
 121                 return True
 122
 123         def recv(self, timeout=120):
 124                 """Attempts to receive a message from the network.
 125
 126                 timeout - max number of seconds to wait for a message.
 127                 If no message is received in 'timeout' seconds, None is returned. """
 128
 129                 msg = None
 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
 136                                 timeout -= endtime
 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()
 140
 141                 # if we've acquired a message, handle it
 142                 if len(self.queue) > 0:
 143                         self.recvCallback(self.queue.pop(0))
 144                 return None
 145
 146
 147