test_loopback.py

#!/usr/local/bin/python
#
#                               Copyright 2004
#                                     by
#                        The Board of Trustees of the
#                     Leland Stanford Junior University.
#                            All rights reserved.
#


__facility__ = "Online"
__abstract__ = "Event loopback test"
__author__   = "A. Perazzo <perazzo@SLAC.Stanford.edu> SLAC - GLAST I&T/Online"
__date__     = ("$Date: 2004/08/24 23:20:56 $").split(' ')[1]
__version__  = "$Revision: 2.1 $"
__release__  = "$Name: R04-12-00 $"
__credits__  = "SLAC"

import LATTE.copyright_SLAC

import threading
import select
import LATTE.client.gLoopback as gLoopback

class EvtCli(object):
  def __init__(self):
    self.__loopback = gLoopback.Loopback()

  def read(self):
    print "readfunction"
    try:
      rd, wr, ex = select.select([self.__loopback.socket()], [], [], None)
      print "select done"
    except Exception, msg:
      print msg
      return
    for fd in rd:
      if fd == self.__loopback.socket():
        value = self.__loopback.acknowledge()
        print "shutdown!", value

  def dontread(self):
    print "dont read called"
    self.__loopback.commit(0xbeef)
    print "dont read done"


if __name__ == '__main__':
  evtcli = EvtCli()
  evtHandler = threading.Thread(None, evtcli.read, 'EventHandler',())
  evtHandler.start()
  evtcli.dontread()
  print "done with threads"

---