examples/ping.py

   1 #!/usr/bin/env python
   2 # -*- coding: utf-8 -*-
   3
   4 """
   5     Slixmpp: The Slick XMPP Library
   6     Copyright (C) 2010  Nathanael C. Fritz
   7     This file is part of Slixmpp.
   8
   9     See the file LICENSE for copying permission.
  10 """
  11
  12 import sys
  13 import logging
  14 import getpass
  15 from optparse import OptionParser
  16
  17 import slixmpp
  18
  19 # Python versions before 3.0 do not use UTF-8 encoding
  20 # by default. To ensure that Unicode is handled properly
  21 # throughout Slixmpp, we will set the default encoding
  22 # ourselves to UTF-8.
  23 if sys.version_info < (3, 0):
  24     from slixmpp.util.misc_ops import setdefaultencoding
  25     setdefaultencoding('utf8')
  26 else:
  27     raw_input = input
  28
  29
  30 class PingTest(slixmpp.ClientXMPP):
  31
  32     """
  33     A simple Slixmpp bot that will send a ping request
  34     to a given JID.
  35     """
  36
  37     def __init__(self, jid, password, pingjid):
  38         slixmpp.ClientXMPP.__init__(self, jid, password)
  39         if pingjid is None:
  40             pingjid = self.boundjid.bare
  41         self.pingjid = pingjid
  42
  43         # The session_start event will be triggered when
  44         # the bot establishes its connection with the server
  45         # and the XML streams are ready for use. We want to
  46         # listen for this event so that we we can initialize
  47         # our roster.
  48         self.add_event_handler("session_start", self.start, threaded=True)
  49
  50     def start(self, event):
  51         """
  52         Process the session_start event.
  53
  54         Typical actions for the session_start event are
  55         requesting the roster and broadcasting an initial
  56         presence stanza.
  57
  58         Arguments:
  59             event -- An empty dictionary. The session_start
  60                      event does not provide any additional
  61                      data.
  62         """
  63         self.send_presence()
  64         self.get_roster()
  65
  66         try:
  67             rtt = self['xep_0199'].ping(self.pingjid,
  68                                         timeout=10)
  69             logging.info("Success! RTT: %s", rtt)
  70         except IqError as e:
  71             logging.info("Error pinging %s: %s",
  72                     self.pingjid,
  73                     e.iq['error']['condition'])
  74         except IqTimeout:
  75             logging.info("No response from %s", self.pingjid)
  76         finally:
  77             self.disconnect()
  78
  79
  80 if __name__ == '__main__':
  81     # Setup the command line arguments.
  82     optp = OptionParser()
  83
  84     # Output verbosity options.
  85     optp.add_option('-q', '--quiet', help='set logging to ERROR',
  86                     action='store_const', dest='loglevel',
  87                     const=logging.ERROR, default=logging.INFO)
  88     optp.add_option('-d', '--debug', help='set logging to DEBUG',
  89                     action='store_const', dest='loglevel',
  90                     const=logging.DEBUG, default=logging.INFO)
  91     optp.add_option('-v', '--verbose', help='set logging to COMM',
  92                     action='store_const', dest='loglevel',
  93                     const=5, default=logging.INFO)
  94     optp.add_option('-t', '--pingto', help='set jid to ping',
  95                     action='store', type='string', dest='pingjid',
  96                     default=None)
  97
  98     # JID and password options.
  99     optp.add_option("-j", "--jid", dest="jid",
 100                     help="JID to use")
 101     optp.add_option("-p", "--password", dest="password",
 102                     help="password to use")
 103
 104     opts, args = optp.parse_args()
 105
 106     # Setup logging.
 107     logging.basicConfig(level=opts.loglevel,
 108                         format='%(levelname)-8s %(message)s')
 109
 110     if opts.jid is None:
 111         opts.jid = raw_input("Username: ")
 112     if opts.password is None:
 113         opts.password = getpass.getpass("Password: ")
 114
 115     # Setup the PingTest and register plugins. Note that while plugins may
 116     # have interdependencies, the order in which you register them does
 117     # not matter.
 118     xmpp = PingTest(opts.jid, opts.password, opts.pingjid)
 119     xmpp.register_plugin('xep_0030')  # Service Discovery
 120     xmpp.register_plugin('xep_0004')  # Data Forms
 121     xmpp.register_plugin('xep_0060')  # PubSub
 122     xmpp.register_plugin('xep_0199')  # XMPP Ping
 123
 124     # If you are working with an OpenFire server, you may need
 125     # to adjust the SSL version used:
 126     # xmpp.ssl_version = ssl.PROTOCOL_SSLv3
 127
 128     # If you want to verify the SSL certificates offered by a server:
 129     # xmpp.ca_certs = "path/to/ca/cert"
 130
 131     # Connect to the XMPP server and start processing XMPP stanzas.
 132     if xmpp.connect():
 133         # If you do not have the dnspython library installed, you will need
 134         # to manually specify the name of the server if it does not match
 135         # the one in the JID. For example, to use Google Talk you would
 136         # need to use:
 137         #
 138         # if xmpp.connect(('talk.google.com', 5222)):
 139         #     ...
 140         xmpp.process(block=True)
 141         print("Done")
 142     else:
 143         print("Unable to connect.")