def testNotifyChange(self):
def change():
hkey = win32api.RegCreateKey(win32con.HKEY_CURRENT_USER, self.key_name)
try:
win32api.RegSetValue(hkey, None, win32con.REG_SZ, "foo")
finally:
win32api.RegDeleteKey(win32con.HKEY_CURRENT_USER, self.key_name)
evt = win32event.CreateEvent(None,0,0,None)
## REG_NOTIFY_CHANGE_LAST_SET - values
## REG_CHANGE_NOTIFY_NAME - keys
## REG_NOTIFY_CHANGE_SECURITY - security descriptor
## REG_NOTIFY_CHANGE_ATTRIBUTES
win32api.RegNotifyChangeKeyValue(win32con.HKEY_CURRENT_USER,1,win32api.REG_NOTIFY_CHANGE_LAST_SET,evt,True)
ret_code=win32event.WaitForSingleObject(evt,0)
# Should be no change.
self.failUnless(ret_code==win32con.WAIT_TIMEOUT)
change()
# Our event should now be in a signalled state.
ret_code=win32event.WaitForSingleObject(evt,0)
self.failUnless(ret_code==win32con.WAIT_OBJECT_0)
python类WAIT_OBJECT_0的实例源码
def testNotifyChange(self):
def change():
hkey = win32api.RegCreateKey(win32con.HKEY_CURRENT_USER, self.key_name)
try:
win32api.RegSetValue(hkey, None, win32con.REG_SZ, "foo")
finally:
win32api.RegDeleteKey(win32con.HKEY_CURRENT_USER, self.key_name)
evt = win32event.CreateEvent(None,0,0,None)
## REG_NOTIFY_CHANGE_LAST_SET - values
## REG_CHANGE_NOTIFY_NAME - keys
## REG_NOTIFY_CHANGE_SECURITY - security descriptor
## REG_NOTIFY_CHANGE_ATTRIBUTES
win32api.RegNotifyChangeKeyValue(win32con.HKEY_CURRENT_USER,1,win32api.REG_NOTIFY_CHANGE_LAST_SET,evt,True)
ret_code=win32event.WaitForSingleObject(evt,0)
# Should be no change.
self.failUnless(ret_code==win32con.WAIT_TIMEOUT)
change()
# Our event should now be in a signalled state.
ret_code=win32event.WaitForSingleObject(evt,0)
self.failUnless(ret_code==win32con.WAIT_OBJECT_0)
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.")
