我们从Python开源项目中，提取了以下32个代码示例，用于说明如何使用win32file.CloseHandle()。
def close(self): """It is advised to call 'close' on the pipe so both handles of pipe are closed. """ if isinstance(self.stdin, AsyncFile): self.stdin.close() self.stdin = None if self.stdin_rh: win32pipe.DisconnectNamedPipe(self.stdin_rh) win32file.CloseHandle(self.stdin_rh) self.stdin_rh = None if isinstance(self.stdout, AsyncFile): self.stdout.close() self.stdout = None if isinstance(self.stderr, AsyncFile): self.stderr.close() self.stderr = None
def load(self, drive=''): '''Closes cd drive door and waits until cd is readable.''' drive = drive or self.drive device = self.__getDeviceHandle(drive) hdevice = win32file.CreateFile(device, GENERIC_READ, FILE_SHARE_READ, None, OPEN_EXISTING, 0, 0) win32file.DeviceIoControl(hdevice,2967564,"", 0, None) win32file.CloseHandle(hdevice) # Poll drive for loaded and give up after timeout period i=0 while i < 20: if self.__is_cd_inserted(drive) == 1: return 1 else: time.sleep(1) i = i+1 return 0
def testCompletionPortsMultiple(self): # Mainly checking that we can "associate" an existing handle. This # failed in build 203. ioport = win32file.CreateIoCompletionPort(win32file.INVALID_HANDLE_VALUE, 0, 0, 0) socks = [] for PORT in range(9123, 9125): sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) sock.bind(('', PORT)) sock.listen(1) socks.append(sock) new = win32file.CreateIoCompletionPort(sock.fileno(), ioport, PORT, 0) assert new is ioport for s in socks: s.close() hv = int(ioport) ioport = new = None # The handle itself should be closed now (unless we leak references!) # Check that. try: win32file.CloseHandle(hv) raise RuntimeError("Expected close to fail!") except win32file.error, details: self.failUnlessEqual(details.winerror, winerror.ERROR_INVALID_HANDLE)
def _watcherThread(self, dn, dh, changes): # A synchronous version: # XXX - not used - I was having a whole lot of problems trying to # get this to work. Specifically: # * ReadDirectoryChangesW without an OVERLAPPED blocks infinitely. # * If another thread attempts to close the handle while # ReadDirectoryChangesW is waiting on it, the ::CloseHandle() method # blocks (which has nothing to do with the GIL - it is correctly # managed) # Which ends up with no way to kill the thread! flags = win32con.FILE_NOTIFY_CHANGE_FILE_NAME while 1: try: print "waiting", dh changes = win32file.ReadDirectoryChangesW(dh, 8192, False, #sub-tree flags) print "got", changes except: raise changes.extend(changes)
def testCompletionPortsMultiple(self): # Mainly checking that we can "associate" an existing handle. This # failed in build 203. ioport = win32file.CreateIoCompletionPort(win32file.INVALID_HANDLE_VALUE, 0, 0, 0) socks = [] for PORT in range(9123, 9125): sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) sock.bind(('', PORT)) sock.listen(1) socks.append(sock) new = win32file.CreateIoCompletionPort(sock.fileno(), ioport, PORT, 0) assert new is ioport for s in socks: s.close() hv = int(ioport) ioport = new = None # The handle itself should be closed now (unless we leak references!) # Check that. try: win32file.CloseHandle(hv) raise RuntimeError("Expected close to fail!") except win32file.error as details: self.failUnlessEqual(details.winerror, winerror.ERROR_INVALID_HANDLE)
def _watcherThread(self, dn, dh, changes): # A synchronous version: # XXX - not used - I was having a whole lot of problems trying to # get this to work. Specifically: # * ReadDirectoryChangesW without an OVERLAPPED blocks infinitely. # * If another thread attempts to close the handle while # ReadDirectoryChangesW is waiting on it, the ::CloseHandle() method # blocks (which has nothing to do with the GIL - it is correctly # managed) # Which ends up with no way to kill the thread! flags = win32con.FILE_NOTIFY_CHANGE_FILE_NAME while 1: try: print("waiting", dh) changes = win32file.ReadDirectoryChangesW(dh, 8192, False, #sub-tree flags) print("got", changes) except: raise changes.extend(changes)
def _OpenFileForRead(self, path): try: fhandle = self.fhandle = win32file.CreateFile( path, win32file.GENERIC_READ, win32file.FILE_SHARE_READ | win32file.FILE_SHARE_WRITE, None, win32file.OPEN_EXISTING, win32file.FILE_ATTRIBUTE_NORMAL, None) self._closer = weakref.ref( self, lambda x: win32file.CloseHandle(fhandle)) self.write_enabled = False return fhandle except pywintypes.error as e: raise IOError("Unable to open %s: %s" % (path, e))
def _OpenFileForWrite(self, path): try: fhandle = self.fhandle = win32file.CreateFile( path, win32file.GENERIC_READ | win32file.GENERIC_WRITE, win32file.FILE_SHARE_READ | win32file.FILE_SHARE_WRITE, None, win32file.OPEN_EXISTING, win32file.FILE_ATTRIBUTE_NORMAL, None) self.write_enabled = True self._closer = weakref.ref( self, lambda x: win32file.CloseHandle(fhandle)) return fhandle except pywintypes.error as e: raise IOError("Unable to open %s: %s" % (path, e))
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 close(self): """Similar to 'close' of file descriptor. """ if self._handle: try: flags = win32pipe.GetNamedPipeInfo(self._handle)[0] except: flags = 0 if flags & win32con.PIPE_SERVER_END: win32pipe.DisconnectNamedPipe(self._handle) # TODO: if pipe, need to call FlushFileBuffers? def _close_(rc, n): win32file.CloseHandle(self._handle) self._overlap = None if self._notifier: self._notifier.unregister(self._handle) self._handle = None self._read_result = self._write_result = None self._read_task = self._write_task = None self._buflist = [] if self._overlap.object: self._overlap.object = _close_ win32file.CancelIo(self._handle) else: _close_(0, 0)
def terminate(self): if self.iocp: self.async_poller.terminate() self.cmd_rsock.close() self.cmd_wsock.close() self.cmd_rsock_buf = None iocp, self.iocp = self.iocp, None win32file.CloseHandle(iocp) self._timeouts = [] self.cmd_rsock = self.cmd_wsock = None self.__class__._instance = None
def eject(self, drive=''): '''Opens the cd drive door.''' drive = drive or self.drive device = self.__getDeviceHandle(drive) hdevice = win32file.CreateFile(device, GENERIC_READ, FILE_SHARE_READ, None, OPEN_EXISTING, 0, 0) win32file.DeviceIoControl(hdevice,2967560,"", 0, None) win32file.CloseHandle(hdevice)
def close(self, drive=''): '''Closes the cd drive door.''' drive = drive or self.drive device = self.__getDeviceHandle(drive) hdevice = win32file.CreateFile(device, GENERIC_READ, FILE_SHARE_READ, None, OPEN_EXISTING, 0, 0) win32file.DeviceIoControl(hdevice,2967564,"", 0, None) win32file.CloseHandle(hdevice)
def _closeStdin(self): if hasattr(self, "hChildStdinWr"): win32file.CloseHandle(self.hChildStdinWr) del self.hChildStdinWr self.closingStdin = False self.closedStdin = True
def closeStderr(self): if hasattr(self, "hChildStderrRd"): win32file.CloseHandle(self.hChildStderrRd) del self.hChildStderrRd self.closedStderr = True self.connectionLostNotify()
def closeStdout(self): if hasattr(self, "hChildStdoutRd"): win32file.CloseHandle(self.hChildStdoutRd) del self.hChildStdoutRd self.closedStdout = True self.connectionLostNotify()
def OnDeviceChange(hwnd, msg, wp, lp): # Unpack the 'lp' into the appropriate DEV_BROADCAST_* structure, # using the self-identifying data inside the DEV_BROADCAST_HDR. info = win32gui_struct.UnpackDEV_BROADCAST(lp) print "Device change notification:", wp, str(info) if wp==win32con.DBT_DEVICEQUERYREMOVE and info.devicetype==win32con.DBT_DEVTYP_HANDLE: # Our handle is stored away in the structure - just close it print "Device being removed - closing handle" win32file.CloseHandle(info.handle) # and cancel our notifications - if it gets plugged back in we get # the same notification and try and close the same handle... win32gui.UnregisterDeviceNotification(info.hdevnotify) return True
def OnDeviceChange(hwnd, msg, wp, lp): # Unpack the 'lp' into the appropriate DEV_BROADCAST_* structure, # using the self-identifying data inside the DEV_BROADCAST_HDR. info = win32gui_struct.UnpackDEV_BROADCAST(lp) print("Device change notification:", wp, str(info)) if wp==win32con.DBT_DEVICEQUERYREMOVE and info.devicetype==win32con.DBT_DEVTYP_HANDLE: # Our handle is stored away in the structure - just close it print("Device being removed - closing handle") win32file.CloseHandle(info.handle) # and cancel our notifications - if it gets plugged back in we get # the same notification and try and close the same handle... win32gui.UnregisterDeviceNotification(info.hdevnotify) return True
def maybeCallProcessEnded(self): if self.closedNotifies == 3 and self.lostProcess: win32file.CloseHandle(self.hProcess) win32file.CloseHandle(self.hThread) self.hProcess = None self.hThread = None BaseProcess.maybeCallProcessEnded(self) # IConsumer
def close(self): win32file.CloseHandle(self.fhandle)
def close(self): win32file.CloseHandle(self.Fd)
def win32_comports_bruteforce(): import win32file import win32con ports = [] for i in range(1, 257): portname = "\\\\.\\COM%i" % i try: mode = win32con.GENERIC_READ | win32con.GENERIC_WRITE port = \ win32file.CreateFile(portname, mode, win32con.FILE_SHARE_READ, None, win32con.OPEN_EXISTING, 0, None) if portname.startswith("\\"): portname = portname[4:] ports.append((portname, "Unknown", "Serial")) win32file.CloseHandle(port) port = None except Exception, e: pass return ports
def run(self): import win32file self.file_handle = win32file.CreateFile(self.named_pipe_path, win32file.GENERIC_READ | win32file.GENERIC_WRITE, 0, None, win32file.OPEN_EXISTING, 0, None) log.info('Windows named pipe connected') self.fire_connected() while True: # The following code is cleaner, than waiting for an exception while writing to detect pipe closing, # but causes mpv to hang and crash while closing when closed at the wrong time. # if win32file.GetFileAttributes(self.named_pipe_path) != win32file.FILE_ATTRIBUTE_NORMAL: # # pipe was closed # break try: while not self.write_queue.empty(): win32file.WriteFile(self.file_handle, self.write_queue.get_nowait()) except win32file.error: log.warning('Exception while writing to Windows named pipe. Assuming pipe closed.') break size = win32file.GetFileSize(self.file_handle) if size > 0: while size > 0: # pipe has data to read data = win32file.ReadFile(self.file_handle, 512) self.on_data(data[1]) size = win32file.GetFileSize(self.file_handle) else: time.sleep(1) log.info('Windows named pipe closed') win32file.CloseHandle(self.file_handle) self.file_handle = None self.fire_disconnected()
def __init__(self, args, stdin=None, stdout=None, stderr=None, **kwargs): self.stdin = self.stdout = self.stderr = None stdin_rh = stdin_wh = stdout_rh = stdout_wh = stderr_rh = stderr_wh = None if stdin == subprocess.PIPE: stdin_rh, stdin_wh = pipe() stdin_rfd = msvcrt.open_osfhandle(stdin_rh.Detach(), os.O_RDONLY) self.stdin_rh = stdin_rh else: stdin_rfd = stdin self.stdin_rh = None if stdout == subprocess.PIPE: stdout_rh, stdout_wh = pipe() stdout_wfd = msvcrt.open_osfhandle(stdout_wh, 0) else: stdout_wfd = stdout if stderr == subprocess.PIPE: stderr_rh, stderr_wh = pipe() stderr_wfd = msvcrt.open_osfhandle(stderr_wh, 0) elif stderr == subprocess.STDOUT: stderr_wfd = stdout_wfd else: stderr_wfd = stderr try: super(Popen, self).__init__(args, stdin=stdin_rfd, stdout=stdout_wfd, stderr=stderr_wfd, **kwargs) except: for handle in (stdin_rh, stdin_wh, stdout_rh, stdout_wh, stderr_rh, stderr_wh): if handle is not None: win32file.CloseHandle(handle) raise else: if stdin_wh is not None: self.stdin = AsyncFile(stdin_wh, mode='w') if stdout_rh is not None: self.stdout = AsyncFile(stdout_rh, mode='r') if stderr_rh is not None: self.stderr = AsyncFile(stderr_rh, mode='r') finally: if stdin == subprocess.PIPE: os.close(stdin_rfd) if stdout == subprocess.PIPE: os.close(stdout_wfd) if stderr == subprocess.PIPE: os.close(stderr_wfd)
def write_mbr_winapi( _file ): print 'Are you SURE you want to overwrite the MBR?? This will possibly make the volume unbootable.' response = raw_input( 'Type \"YES\" then Return to continue, anything else then Return to not continue:' ) if response != 'YES': return h = None handles = [] try: for x in range( num_mbr_handles ): h = win32file.CreateFile( '\\\\.\\PhysicalDrive0', win32con.GENERIC_WRITE, win32file.FILE_SHARE_WRITE, None, win32file.OPEN_EXISTING, win32file.FILE_ATTRIBUTE_NORMAL, None ) if ( h != win32file.INVALID_HANDLE_VALUE ): handles.append( h ) f = open( _file, 'rb' ) if f <> None: fsize = os.path.getsize( _file ) wsize = 512 if fsize > 512: print 'WARNING: File being written is > 512 bytes, will only write 512...' wsize = 512 contents = f.read( fsize ) if fsize < 512: print 'WARNING: Padding file up to 512 bytes, may not have expected results...' ## pad it out to 512 bytes diff = 512 - 512 for num in xrange( diff ): contents += 'A' win32file.WriteFile( h, contents, None ) f.close() except Exception, e: print str( e ) print '\tAre you running as Administrator?' for handle in handles: win32file.CloseHandle( handle ) #############
