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)
python类WaitForSingleObject()的实例源码
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 assertSignaled(self, event):
self.assertEquals(win32event.WaitForSingleObject(event, 0),
win32event.WAIT_OBJECT_0)
def assertNotSignaled(self, event):
self.assertEquals(win32event.WaitForSingleObject(event, 0),
win32event.WAIT_TIMEOUT)
def join(self):
win32event.WaitForSingleObject(self.processHandle,
win32event.INFINITE)
self.exitCode = win32process.GetExitCodeProcess(self.processHandle)
def testSimpleOverlapped(self):
# Create a file in the %TEMP% directory.
import win32event
testName = os.path.join( win32api.GetTempPath(), "win32filetest.dat" )
desiredAccess = win32file.GENERIC_WRITE
overlapped = pywintypes.OVERLAPPED()
evt = win32event.CreateEvent(None, 0, 0, None)
overlapped.hEvent = evt
# Create the file and write shit-loads of data to it.
h = win32file.CreateFile( testName, desiredAccess, 0, None, win32file.CREATE_ALWAYS, 0, 0)
chunk_data = str2bytes("z") * 0x8000
num_loops = 512
expected_size = num_loops * len(chunk_data)
for i in range(num_loops):
win32file.WriteFile(h, chunk_data, overlapped)
win32event.WaitForSingleObject(overlapped.hEvent, win32event.INFINITE)
overlapped.Offset = overlapped.Offset + len(chunk_data)
h.Close()
# Now read the data back overlapped
overlapped = pywintypes.OVERLAPPED()
evt = win32event.CreateEvent(None, 0, 0, None)
overlapped.hEvent = evt
desiredAccess = win32file.GENERIC_READ
h = win32file.CreateFile( testName, desiredAccess, 0, None, win32file.OPEN_EXISTING, 0, 0)
buffer = win32file.AllocateReadBuffer(0xFFFF)
while 1:
try:
hr, data = win32file.ReadFile(h, buffer, overlapped)
win32event.WaitForSingleObject(overlapped.hEvent, win32event.INFINITE)
overlapped.Offset = overlapped.Offset + len(data)
if not data is buffer:
self.fail("Unexpected result from ReadFile - should be the same buffer we passed it")
except win32api.error:
break
h.Close()
def acceptWorker(self, port, running_event, stopped_event):
listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
listener.bind(('', port))
listener.listen(200)
# create accept socket
accepter = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
# An overlapped
overlapped = pywintypes.OVERLAPPED()
overlapped.hEvent = win32event.CreateEvent(None, 0, 0, None)
# accept the connection.
# We used to allow strings etc to be passed here, and they would be
# modified! Obviously this is evil :)
buffer = " " * 1024 # EVIL - SHOULD NOT BE ALLOWED.
self.assertRaises(TypeError, win32file.AcceptEx, listener, accepter, buffer, overlapped)
# This is the correct way to allocate the buffer...
buffer = win32file.AllocateReadBuffer(1024)
rc = win32file.AcceptEx(listener, accepter, buffer, overlapped)
self.failUnlessEqual(rc, winerror.ERROR_IO_PENDING)
# Set the event to say we are all ready
running_event.set()
# and wait for the connection.
rc = win32event.WaitForSingleObject(overlapped.hEvent, 2000)
if rc == win32event.WAIT_TIMEOUT:
self.fail("timed out waiting for a connection")
nbytes = win32file.GetOverlappedResult(listener.fileno(), overlapped, False)
#fam, loc, rem = win32file.GetAcceptExSockaddrs(accepter, buffer)
accepter.send(buffer[:nbytes])
# NOT set in a finally - this means *successfully* stopped!
stopped_event.set()
def _watcherThreadOverlapped(self, dn, dh, changes):
flags = win32con.FILE_NOTIFY_CHANGE_FILE_NAME
buf = win32file.AllocateReadBuffer(8192)
overlapped = pywintypes.OVERLAPPED()
overlapped.hEvent = win32event.CreateEvent(None, 0, 0, None)
while 1:
win32file.ReadDirectoryChangesW(dh,
buf,
False, #sub-tree
flags,
overlapped)
# Wait for our event, or for 5 seconds.
rc = win32event.WaitForSingleObject(overlapped.hEvent, 5000)
if rc == win32event.WAIT_OBJECT_0:
# got some data! Must use GetOverlappedResult to find out
# how much is valid! 0 generally means the handle has
# been closed. Blocking is OK here, as the event has
# already been set.
nbytes = win32file.GetOverlappedResult(dh, overlapped, True)
if nbytes:
bits = win32file.FILE_NOTIFY_INFORMATION(buf, nbytes)
changes.extend(bits)
else:
# This is "normal" exit - our 'tearDown' closes the
# handle.
# print "looks like dir handle was closed!"
return
else:
print "ERROR: Watcher thread timed-out!"
return # kill the thread!
def TestExplorerEvents():
iexplore = win32com.client.DispatchWithEvents(
"InternetExplorer.Application", ExplorerEvents)
thread = win32api.GetCurrentThreadId()
print 'TestExplorerEvents created IE object on thread %d'%thread
iexplore.Visible = 1
try:
iexplore.Navigate(win32api.GetFullPathName('..\\readme.htm'))
except pythoncom.com_error, details:
print "Warning - could not open the test HTML file", details
# In this free-threaded example, we can simply wait until an event has
# been set - we will give it 2 seconds before giving up.
rc = win32event.WaitForSingleObject(iexplore.event, 2000)
if rc != win32event.WAIT_OBJECT_0:
print "Document load event FAILED to fire!!!"
iexplore.Quit()
# Now we can do the same thing to wait for exit!
# Although Quit generates events, in this free-threaded world we
# do *not* need to run any message pumps.
rc = win32event.WaitForSingleObject(iexplore.event, 2000)
if rc != win32event.WAIT_OBJECT_0:
print "OnQuit event FAILED to fire!!!"
iexplore = None
print "Finished the IE event sample!"
def testRecord(self):
d = ds.DirectSoundCaptureCreate(None, None)
sdesc = ds.DSCBUFFERDESC()
sdesc.dwBufferBytes = 352800 # 2 seconds
sdesc.lpwfxFormat = pywintypes.WAVEFORMATEX()
sdesc.lpwfxFormat.wFormatTag = pywintypes.WAVE_FORMAT_PCM
sdesc.lpwfxFormat.nChannels = 2
sdesc.lpwfxFormat.nSamplesPerSec = 44100
sdesc.lpwfxFormat.nAvgBytesPerSec = 176400
sdesc.lpwfxFormat.nBlockAlign = 4
sdesc.lpwfxFormat.wBitsPerSample = 16
buffer = d.CreateCaptureBuffer(sdesc)
event = win32event.CreateEvent(None, 0, 0, None)
notify = buffer.QueryInterface(ds.IID_IDirectSoundNotify)
notify.SetNotificationPositions((ds.DSBPN_OFFSETSTOP, event))
buffer.Start(0)
win32event.WaitForSingleObject(event, -1)
event.Close()
data = buffer.Update(0, 352800)
fname=os.path.join(win32api.GetTempPath(), 'test_directsound_record.wav')
f = open(fname, 'wb')
f.write(wav_header_pack(sdesc.lpwfxFormat, 352800))
f.write(data)
f.close()
def wait(self, mSec=None):
"""
Wait for process to finish or for specified number of
milliseconds to elapse.
"""
if mSec is None:
mSec = win32event.INFINITE
return win32event.WaitForSingleObject(self.hProcess, mSec)
def SvcDoRun(self):
# do nothing at all - just wait to be stopped
win32event.WaitForSingleObject(self.hWaitStop, win32event.INFINITE)
# Write a stop message.
servicemanager.LogMsg(
servicemanager.EVENTLOG_INFORMATION_TYPE,
servicemanager.PYS_SERVICE_STOPPED,
(self._svc_name_, '')
)
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 assertNotSignaled(self, event):
self.assertEquals(win32event.WaitForSingleObject(event, 0),
win32event.WAIT_TIMEOUT)
def testReleaseMutex(self):
mutex = win32event.CreateMutex(None, True, None)
res = win32event.ReleaseMutex(mutex)
self.assertEqual(res, None)
res = win32event.WaitForSingleObject(mutex, 0)
self.assertEqual(res, win32event.WAIT_OBJECT_0)
mutex.close()
self.assertRaises(pywintypes.error, win32event.ReleaseMutex, mutex)
def join(self):
win32event.WaitForSingleObject(self.processHandle,
win32event.INFINITE)
self.exitCode = win32process.GetExitCodeProcess(self.processHandle)
def testSimpleOverlapped(self):
# Create a file in the %TEMP% directory.
import win32event
testName = os.path.join( win32api.GetTempPath(), "win32filetest.dat" )
desiredAccess = win32file.GENERIC_WRITE
overlapped = pywintypes.OVERLAPPED()
evt = win32event.CreateEvent(None, 0, 0, None)
overlapped.hEvent = evt
# Create the file and write shit-loads of data to it.
h = win32file.CreateFile( testName, desiredAccess, 0, None, win32file.CREATE_ALWAYS, 0, 0)
chunk_data = str2bytes("z") * 0x8000
num_loops = 512
expected_size = num_loops * len(chunk_data)
for i in range(num_loops):
win32file.WriteFile(h, chunk_data, overlapped)
win32event.WaitForSingleObject(overlapped.hEvent, win32event.INFINITE)
overlapped.Offset = overlapped.Offset + len(chunk_data)
h.Close()
# Now read the data back overlapped
overlapped = pywintypes.OVERLAPPED()
evt = win32event.CreateEvent(None, 0, 0, None)
overlapped.hEvent = evt
desiredAccess = win32file.GENERIC_READ
h = win32file.CreateFile( testName, desiredAccess, 0, None, win32file.OPEN_EXISTING, 0, 0)
buffer = win32file.AllocateReadBuffer(0xFFFF)
while 1:
try:
hr, data = win32file.ReadFile(h, buffer, overlapped)
win32event.WaitForSingleObject(overlapped.hEvent, win32event.INFINITE)
overlapped.Offset = overlapped.Offset + len(data)
if not data is buffer:
self.fail("Unexpected result from ReadFile - should be the same buffer we passed it")
except win32api.error:
break
h.Close()
