@_FileLikeObjectBase._before_close
@retry_method
def decompress(self, decompress, size, num_retries=None):
- for segment in self.readall(size, num_retries):
+ for segment in self.readall(size, num_retries=num_retries):
data = decompress(segment)
if data:
yield data
else:
raise AssertionError("Buffer block is not writable")
+ STATE_TRANSITIONS = frozenset([
+ (WRITABLE, PENDING),
+ (PENDING, COMMITTED),
+ (PENDING, ERROR),
+ (ERROR, PENDING)])
+
@synchronized
def set_state(self, nextstate, val=None):
- if ((self._state == _BufferBlock.WRITABLE and nextstate == _BufferBlock.PENDING) or
- (self._state == _BufferBlock.PENDING and nextstate == _BufferBlock.COMMITTED) or
- (self._state == _BufferBlock.PENDING and nextstate == _BufferBlock.ERROR) or
- (self._state == _BufferBlock.ERROR and nextstate == _BufferBlock.PENDING)):
- self._state = nextstate
-
- if self._state == _BufferBlock.PENDING:
- self.wait_for_commit.clear()
-
- if self._state == _BufferBlock.COMMITTED:
- self._locator = val
- self.buffer_view = None
- self.buffer_block = None
- self.wait_for_commit.set()
-
- if self._state == _BufferBlock.ERROR:
- self.error = val
- self.wait_for_commit.set()
- else:
+ if (self._state, nextstate) not in self.STATE_TRANSITIONS:
raise StateChangeError("Invalid state change from %s to %s" % (self.state, nextstate), self.state, nextstate)
+ self._state = nextstate
+
+ if self._state == _BufferBlock.PENDING:
+ self.wait_for_commit.clear()
+
+ if self._state == _BufferBlock.COMMITTED:
+ self._locator = val
+ self.buffer_view = None
+ self.buffer_block = None
+ self.wait_for_commit.set()
+
+ if self._state == _BufferBlock.ERROR:
+ self.error = val
+ self.wait_for_commit.set()
@synchronized
def state(self):
@retry_method
def writelines(self, seq, num_retries=None):
for s in seq:
- self.write(s, num_retries)
+ self.write(s, num_retries=num_retries)
@_FileLikeObjectBase._before_close
def truncate(self, size=None):