def setUp(self):
self.watcher_threads = []
self.watcher_thread_changes = []
self.dir_names = []
self.dir_handles = []
for i in range(self.num_test_dirs):
td = tempfile.mktemp("-test-directory-changes-%d" % i)
os.mkdir(td)
self.dir_names.append(td)
hdir = win32file.CreateFile(td,
ntsecuritycon.FILE_LIST_DIRECTORY,
win32con.FILE_SHARE_READ,
None, # security desc
win32con.OPEN_EXISTING,
win32con.FILE_FLAG_BACKUP_SEMANTICS |
win32con.FILE_FLAG_OVERLAPPED,
None)
self.dir_handles.append(hdir)
changes = []
t = threading.Thread(target=self._watcherThreadOverlapped,
args=(td, hdir, changes))
t.start()
self.watcher_threads.append(t)
self.watcher_thread_changes.append(changes)
python类FILE_FLAG_OVERLAPPED的实例源码
def setUp(self):
self.watcher_threads = []
self.watcher_thread_changes = []
self.dir_names = []
self.dir_handles = []
for i in range(self.num_test_dirs):
td = tempfile.mktemp("-test-directory-changes-%d" % i)
os.mkdir(td)
self.dir_names.append(td)
hdir = win32file.CreateFile(td,
ntsecuritycon.FILE_LIST_DIRECTORY,
win32con.FILE_SHARE_READ,
None, # security desc
win32con.OPEN_EXISTING,
win32con.FILE_FLAG_BACKUP_SEMANTICS |
win32con.FILE_FLAG_OVERLAPPED,
None)
self.dir_handles.append(hdir)
changes = []
t = threading.Thread(target=self._watcherThreadOverlapped,
args=(td, hdir, changes))
t.start()
self.watcher_threads.append(t)
self.watcher_thread_changes.append(changes)
def pipe(bufsize=8192):
"""Creates overlapped (asynchronous) pipe.
"""
name = r'\\.\pipe\pycos-pipe-%d-%d' % (os.getpid(), next(_pipe_id))
openmode = (win32pipe.PIPE_ACCESS_INBOUND | win32file.FILE_FLAG_OVERLAPPED |
FILE_FLAG_FIRST_PIPE_INSTANCE)
pipemode = (win32pipe.PIPE_TYPE_BYTE | win32pipe.PIPE_READMODE_BYTE)
rh = wh = None
try:
rh = win32pipe.CreateNamedPipe(
name, openmode, pipemode, 1, bufsize, bufsize,
win32pipe.NMPWAIT_USE_DEFAULT_WAIT, None)
wh = win32file.CreateFile(
name, win32file.GENERIC_WRITE | winnt.FILE_READ_ATTRIBUTES, 0, None,
win32file.OPEN_EXISTING, win32file.FILE_FLAG_OVERLAPPED, None)
overlapped = pywintypes.OVERLAPPED()
# 'yield' can't be used in constructor so use sync wait
# (in this case it is should be okay)
overlapped.hEvent = win32event.CreateEvent(None, 0, 0, None)
rc = win32pipe.ConnectNamedPipe(rh, overlapped)
if rc == winerror.ERROR_PIPE_CONNECTED:
win32event.SetEvent(overlapped.hEvent)
rc = win32event.WaitForSingleObject(overlapped.hEvent, 1000)
overlapped = None
if rc != win32event.WAIT_OBJECT_0:
pycos.logger.warning('connect failed: %s' % rc)
raise Exception(rc)
return (rh, wh)
except:
if rh is not None:
win32file.CloseHandle(rh)
if wh is not None:
win32file.CloseHandle(wh)
raise
def pipe(bufsize=8192):
"""Creates overlapped (asynchronous) pipe.
"""
name = r'\\.\pipe\pycos-pipe-%d-%d' % (os.getpid(), next(_pipe_id))
openmode = (win32pipe.PIPE_ACCESS_INBOUND | win32file.FILE_FLAG_OVERLAPPED |
FILE_FLAG_FIRST_PIPE_INSTANCE)
pipemode = (win32pipe.PIPE_TYPE_BYTE | win32pipe.PIPE_READMODE_BYTE)
rh = wh = None
try:
rh = win32pipe.CreateNamedPipe(
name, openmode, pipemode, 1, bufsize, bufsize,
win32pipe.NMPWAIT_USE_DEFAULT_WAIT, None)
wh = win32file.CreateFile(
name, win32file.GENERIC_WRITE | winnt.FILE_READ_ATTRIBUTES, 0, None,
win32file.OPEN_EXISTING, win32file.FILE_FLAG_OVERLAPPED, None)
overlapped = pywintypes.OVERLAPPED()
# 'yield' can't be used in constructor so use sync wait
# (in this case it is should be okay)
overlapped.hEvent = win32event.CreateEvent(None, 0, 0, None)
rc = win32pipe.ConnectNamedPipe(rh, overlapped)
if rc == winerror.ERROR_PIPE_CONNECTED:
win32event.SetEvent(overlapped.hEvent)
rc = win32event.WaitForSingleObject(overlapped.hEvent, 1000)
overlapped = None
if rc != win32event.WAIT_OBJECT_0:
pycos.logger.warning('connect failed: %s' % rc)
raise Exception(rc)
return (rh, wh)
except:
if rh is not None:
win32file.CloseHandle(rh)
if wh is not None:
win32file.CloseHandle(wh)
raise
def create_pipe(pipeName):
# XXX - We should validate here and return None on fail
pipe = win32pipe.CreateNamedPipe(pipeName,
win32pipe.PIPE_ACCESS_INBOUND | win32file.FILE_FLAG_OVERLAPPED,
0x0,
1,
1024,
1024,
0,
None)
return pipe
def testCompletionPortsNonQueued(self, test_overlapped_death = 0):
# In 204 we had a reference count bug when OVERLAPPED objects were
# associated with a completion port other than via
# PostQueuedCompletionStatus. This test is based on the reproduction
# reported with that bug.
# Create the pipe.
BUFSIZE = 512
pipe_name = r"\\.\pipe\pywin32_test_pipe"
handle = win32pipe.CreateNamedPipe(pipe_name,
win32pipe.PIPE_ACCESS_DUPLEX|
win32file.FILE_FLAG_OVERLAPPED,
win32pipe.PIPE_TYPE_MESSAGE|
win32pipe.PIPE_READMODE_MESSAGE|
win32pipe.PIPE_WAIT,
1, BUFSIZE, BUFSIZE,
win32pipe.NMPWAIT_WAIT_FOREVER,
None)
# Create an IOCP and associate it with the handle.
port = win32file.CreateIoCompletionPort(-1, 0, 0, 0)
win32file.CreateIoCompletionPort(handle, port, 1, 0)
t = threading.Thread(target=self._IOCPServerThread, args=(handle,port, test_overlapped_death))
t.setDaemon(True) # avoid hanging entire test suite on failure.
t.start()
try:
time.sleep(0.1) # let thread do its thing.
try:
win32pipe.CallNamedPipe(r"\\.\pipe\pywin32_test_pipe", str2bytes("Hello there"), BUFSIZE, 0)
except win32pipe.error:
# Testing for overlapped death causes this
if not test_overlapped_death:
raise
finally:
if not test_overlapped_death:
handle.Close()
t.join(3)
self.failIf(t.isAlive(), "thread didn't finish")
def testCompletionPortsNonQueued(self, test_overlapped_death = 0):
# In 204 we had a reference count bug when OVERLAPPED objects were
# associated with a completion port other than via
# PostQueuedCompletionStatus. This test is based on the reproduction
# reported with that bug.
# Create the pipe.
BUFSIZE = 512
pipe_name = r"\\.\pipe\pywin32_test_pipe"
handle = win32pipe.CreateNamedPipe(pipe_name,
win32pipe.PIPE_ACCESS_DUPLEX|
win32file.FILE_FLAG_OVERLAPPED,
win32pipe.PIPE_TYPE_MESSAGE|
win32pipe.PIPE_READMODE_MESSAGE|
win32pipe.PIPE_WAIT,
1, BUFSIZE, BUFSIZE,
win32pipe.NMPWAIT_WAIT_FOREVER,
None)
# Create an IOCP and associate it with the handle.
port = win32file.CreateIoCompletionPort(-1, 0, 0, 0)
win32file.CreateIoCompletionPort(handle, port, 1, 0)
t = threading.Thread(target=self._IOCPServerThread, args=(handle,port, test_overlapped_death))
t.setDaemon(True) # avoid hanging entire test suite on failure.
t.start()
try:
time.sleep(0.1) # let thread do its thing.
try:
win32pipe.CallNamedPipe(r"\\.\pipe\pywin32_test_pipe", str2bytes("Hello there"), BUFSIZE, 0)
except win32pipe.error:
# Testing for overlapped death causes this
if not test_overlapped_death:
raise
finally:
if not test_overlapped_death:
handle.Close()
t.join(3)
self.failIf(t.isAlive(), "thread didn't finish")
def create_pipe(pipeName):
# XXX - We should validate here and return None on fail
pipe = win32pipe.CreateNamedPipe(pipeName,
win32pipe.PIPE_ACCESS_INBOUND | win32file.FILE_FLAG_OVERLAPPED,
0x0,
1,
1024,
1024,
0,
None)
return pipe
