mercurial/windows.py

43505:47fac1692ede

    winreg.CloseKey
except ImportError:
    import winreg

osutil = policy.importmod(r'osutil')

getfsmountpoint = win32.getvolumename
getfstype = win32.getfstype
getuser = win32.getuser
hidewindow = win32.hidewindow

    OPNONE = 0
    OPREAD = 1
    OPWRITE = 2

    def __init__(self, fp):
        object.__setattr__(self, r'_fp', fp)
        object.__setattr__(self, r'_lastop', 0)

    def __enter__(self):
        self._fp.__enter__()
        return self

    def _noopseek(self):
        self._fp.seek(0, os.SEEK_CUR)

    def seek(self, *args, **kwargs):
        object.__setattr__(self, r'_lastop', self.OPNONE)
        return self._fp.seek(*args, **kwargs)

    def write(self, d):
        if self._lastop == self.OPREAD:
            self._noopseek()

        object.__setattr__(self, r'_lastop', self.OPWRITE)
        return self._fp.write(d)

    def writelines(self, *args, **kwargs):
        if self._lastop == self.OPREAD:
            self._noopeseek()

        object.__setattr__(self, r'_lastop', self.OPWRITE)
        return self._fp.writelines(*args, **kwargs)

    def read(self, *args, **kwargs):
        if self._lastop == self.OPWRITE:
            self._noopseek()

        object.__setattr__(self, r'_lastop', self.OPREAD)
        return self._fp.read(*args, **kwargs)

    def readline(self, *args, **kwargs):
        if self._lastop == self.OPWRITE:
            self._noopseek()

        object.__setattr__(self, r'_lastop', self.OPREAD)
        return self._fp.readline(*args, **kwargs)

    def readlines(self, *args, **kwargs):
        if self._lastop == self.OPWRITE:
            self._noopseek()

        object.__setattr__(self, r'_lastop', self.OPREAD)
        return self._fp.readlines(*args, **kwargs)


class fdproxy(object):
    """Wraps osutil.posixfile() to override the name attribute to reflect the

        return fp
    except WindowsError as err:
        # convert to a friendlier exception
        raise IOError(
            err.errno, r'%s: %s' % (encoding.strfromlocal(name), err.strerror)
        )


# may be wrapped by win32mbcs extension
listdir = osutil.listdir

                start = end
        except IOError as inst:
            if inst.errno != 0 and not win32.lasterrorwaspipeerror(inst):
                raise
            self.close()
            raise IOError(errno.EPIPE, r'Broken pipe')

    def flush(self):
        try:
            return self.fp.flush()
        except IOError as inst:
            if not win32.lasterrorwaspipeerror(inst):
                raise
            raise IOError(errno.EPIPE, r'Broken pipe')


def _is_win_9x():
    '''return true if run on windows 95, 98 or me.'''
    try:

    return b''.join(chunks)


def bindunixsocket(sock, path):
    raise NotImplementedError(r'unsupported platform')

43506:9f70512ae2cf

    winreg.CloseKey
except ImportError:
    import winreg

osutil = policy.importmod('osutil')

getfsmountpoint = win32.getvolumename
getfstype = win32.getfstype
getuser = win32.getuser
hidewindow = win32.hidewindow

    OPNONE = 0
    OPREAD = 1
    OPWRITE = 2

    def __init__(self, fp):
        object.__setattr__(self, '_fp', fp)
        object.__setattr__(self, '_lastop', 0)

    def __enter__(self):
        self._fp.__enter__()
        return self

    def _noopseek(self):
        self._fp.seek(0, os.SEEK_CUR)

    def seek(self, *args, **kwargs):
        object.__setattr__(self, '_lastop', self.OPNONE)
        return self._fp.seek(*args, **kwargs)

    def write(self, d):
        if self._lastop == self.OPREAD:
            self._noopseek()

        object.__setattr__(self, '_lastop', self.OPWRITE)
        return self._fp.write(d)

    def writelines(self, *args, **kwargs):
        if self._lastop == self.OPREAD:
            self._noopeseek()

        object.__setattr__(self, '_lastop', self.OPWRITE)
        return self._fp.writelines(*args, **kwargs)

    def read(self, *args, **kwargs):
        if self._lastop == self.OPWRITE:
            self._noopseek()

        object.__setattr__(self, '_lastop', self.OPREAD)
        return self._fp.read(*args, **kwargs)

    def readline(self, *args, **kwargs):
        if self._lastop == self.OPWRITE:
            self._noopseek()

        object.__setattr__(self, '_lastop', self.OPREAD)
        return self._fp.readline(*args, **kwargs)

    def readlines(self, *args, **kwargs):
        if self._lastop == self.OPWRITE:
            self._noopseek()

        object.__setattr__(self, '_lastop', self.OPREAD)
        return self._fp.readlines(*args, **kwargs)


class fdproxy(object):
    """Wraps osutil.posixfile() to override the name attribute to reflect the

        return fp
    except WindowsError as err:
        # convert to a friendlier exception
        raise IOError(
            err.errno, '%s: %s' % (encoding.strfromlocal(name), err.strerror)
        )


# may be wrapped by win32mbcs extension
listdir = osutil.listdir

                start = end
        except IOError as inst:
            if inst.errno != 0 and not win32.lasterrorwaspipeerror(inst):
                raise
            self.close()
            raise IOError(errno.EPIPE, 'Broken pipe')

    def flush(self):
        try:
            return self.fp.flush()
        except IOError as inst:
            if not win32.lasterrorwaspipeerror(inst):
                raise
            raise IOError(errno.EPIPE, 'Broken pipe')


def _is_win_9x():
    '''return true if run on windows 95, 98 or me.'''
    try:

    return b''.join(chunks)


def bindunixsocket(sock, path):
    raise NotImplementedError('unsupported platform')