Python win32event 模块，CreateWaitableTimer() 实例源码

def pacemaker(self, timeout=60):
    #   This is a stand-alone heartbeat generator.  To pulse from your own control loop,
    #   call your AbstractLog subclass instance event handler (e.g. AbstractLog['event']()
        def __target(timeout=60):
            if platform.uname()[0].lower() == "windows":
                import win32con
                import win32event
                self.running = True
                kill = win32event.CreateEvent(None, 1, 0, None)
                pulse = win32event.CreateWaitableTimer(None, 0, None)
                win32event.SetWaitableTimer(pulse, 0, timeout*1000, None, None, False)
                while(self.running):
                    try:
                        result = win32event.WaitForMultipleObjects([kill, pulse], False, 1000)

                        # if kill signal received, break loop
                        if(result == win32con.WAIT_OBJECT_0): break
                        # elif timeout has passed, generate a pulse
                        elif(result == win32con.WAIT_OBJECT_0 + 1): self['event']()
                    except:
                        self.notifyOfError("Pacemaker shutdown.  Heartbeats will not be generated.")
                        win32event.SetEvent(kill)
            elif self.options['Verbose']: print "Pacemaker only supported in Windows at this time. " 

        try:
            self.thread = threading.Thread(target=__target, args=(timeout,) )
            self.thread.start()
        except:
            self.notifyOfError("Pacemaker thread exception.  Heartbeats will not be generated.")

def testWaitableFireLong(self):
        h = win32event.CreateWaitableTimer(None, 0, None)
        dt = int2long(-160) # 160 ns.
        win32event.SetWaitableTimer(h, dt, 0, None, None, 0)
        rc = win32event.WaitForSingleObject(h, 1000)
        self.failUnlessEqual(rc, win32event.WAIT_OBJECT_0)

def testWaitableFire(self):
        h = win32event.CreateWaitableTimer(None, 0, None)
        dt = -160 # 160 ns.
        win32event.SetWaitableTimer(h, dt, 0, None, None, 0)
        rc = win32event.WaitForSingleObject(h, 1000)
        self.failUnlessEqual(rc, win32event.WAIT_OBJECT_0)

def testWaitableTrigger(self):
        h = win32event.CreateWaitableTimer(None, 0, None)
        # for the sake of this, pass a long that doesn't fit in an int.
        dt = -2000000000
        win32event.SetWaitableTimer(h, dt, 0, None, None, 0)
        rc = win32event.WaitForSingleObject(h, 10) # 10 ms.
        self.failUnlessEqual(rc, win32event.WAIT_TIMEOUT)

def testWaitableError(self):
        h = win32event.CreateWaitableTimer(None, 0, None)
        h.close()
        self.assertRaises(pywintypes.error, win32event.SetWaitableTimer,
                          h, -42, 0, None, None, 0)

def testWaitableFireLong(self):
        h = win32event.CreateWaitableTimer(None, 0, None)
        dt = int2long(-160) # 160 ns.
        win32event.SetWaitableTimer(h, dt, 0, None, None, 0)
        rc = win32event.WaitForSingleObject(h, 1000)
        self.failUnlessEqual(rc, win32event.WAIT_OBJECT_0)

def testWaitableFire(self):
        h = win32event.CreateWaitableTimer(None, 0, None)
        dt = -160 # 160 ns.
        win32event.SetWaitableTimer(h, dt, 0, None, None, 0)
        rc = win32event.WaitForSingleObject(h, 1000)
        self.failUnlessEqual(rc, win32event.WAIT_OBJECT_0)

def testWaitableTrigger(self):
        h = win32event.CreateWaitableTimer(None, 0, None)
        # for the sake of this, pass a long that doesn't fit in an int.
        dt = -2000000000
        win32event.SetWaitableTimer(h, dt, 0, None, None, 0)
        rc = win32event.WaitForSingleObject(h, 10) # 10 ms.
        self.failUnlessEqual(rc, win32event.WAIT_TIMEOUT)

def testWaitableError(self):
        h = win32event.CreateWaitableTimer(None, 0, None)
        h.close()
        self.assertRaises(pywintypes.error, win32event.SetWaitableTimer,
                          h, -42, 0, None, None, 0)