mirror of
https://github.com/facebook/sapling.git
synced 2024-10-07 15:27:13 +03:00
bundle2: add an unbundle part responsible from unbundling part
We have a new unbundle class and it is now responsible from extracting its own data. The top level bundler only extracts the header (to detect an end of stream marker) then leaves everything else to the `unbundlepart` class. The ultimate goal is to have `unbundlepart` responsible for lazily extracting its payload. This is mostly code movement.
This commit is contained in:
Pierre-Yves David
parent
10657d6382
commit
0819f89a04
@ -448,71 +448,22 @@ class unbundle20(unpackermixin):
|
||||
# make sure param have been loaded
|
||||
self.params
|
||||
self.ui.debug('start extraction of bundle2 parts\n')
|
||||
part = self._readpart()
|
||||
while part is not None:
|
||||
headerblock = self._readpartheader()
|
||||
while headerblock is not None:
|
||||
part = unbundlepart(self.ui, headerblock, self._fp)
|
||||
yield part
|
||||
part = self._readpart()
|
||||
headerblock = self._readpartheader()
|
||||
self.ui.debug('end of bundle2 stream\n')
|
||||
|
||||
def _readpart(self):
|
||||
"""return None when an end of stream markers is reach"""
|
||||
def _readpartheader(self):
|
||||
"""reads a part header size and return the bytes blob
|
||||
|
||||
returns None if empty"""
|
||||
headersize = self._unpack(_fpartheadersize)[0]
|
||||
self.ui.debug('part header size: %i\n' % headersize)
|
||||
if not headersize:
|
||||
return None
|
||||
headerblock = self._readexact(headersize)
|
||||
# some utility to help reading from the header block
|
||||
self._offset = 0 # layer violation to have something easy to understand
|
||||
def fromheader(size):
|
||||
"""return the next <size> byte from the header"""
|
||||
offset = self._offset
|
||||
data = headerblock[offset:(offset + size)]
|
||||
self._offset = offset + size
|
||||
return data
|
||||
def unpackheader(format):
|
||||
"""read given format from header
|
||||
|
||||
This automatically compute the size of the format to read."""
|
||||
data = fromheader(struct.calcsize(format))
|
||||
return _unpack(format, data)
|
||||
|
||||
typesize = unpackheader(_fparttypesize)[0]
|
||||
parttype = fromheader(typesize)
|
||||
self.ui.debug('part type: "%s"\n' % parttype)
|
||||
partid = unpackheader(_fpartid)[0]
|
||||
self.ui.debug('part id: "%s"\n' % partid)
|
||||
## reading parameters
|
||||
# param count
|
||||
mancount, advcount = unpackheader(_fpartparamcount)
|
||||
self.ui.debug('part parameters: %i\n' % (mancount + advcount))
|
||||
# param size
|
||||
paramsizes = unpackheader(_makefpartparamsizes(mancount + advcount))
|
||||
# make it a list of couple again
|
||||
paramsizes = zip(paramsizes[::2], paramsizes[1::2])
|
||||
# split mandatory from advisory
|
||||
mansizes = paramsizes[:mancount]
|
||||
advsizes = paramsizes[mancount:]
|
||||
# retrive param value
|
||||
manparams = []
|
||||
for key, value in mansizes:
|
||||
manparams.append((fromheader(key), fromheader(value)))
|
||||
advparams = []
|
||||
for key, value in advsizes:
|
||||
advparams.append((fromheader(key), fromheader(value)))
|
||||
del self._offset # clean up layer, nobody saw anything.
|
||||
## part payload
|
||||
payload = []
|
||||
payloadsize = self._unpack(_fpayloadsize)[0]
|
||||
self.ui.debug('payload chunk size: %i\n' % payloadsize)
|
||||
while payloadsize:
|
||||
payload.append(self._readexact(payloadsize))
|
||||
payloadsize = self._unpack(_fpayloadsize)[0]
|
||||
self.ui.debug('payload chunk size: %i\n' % payloadsize)
|
||||
payload = ''.join(payload)
|
||||
current = bundlepart(parttype, manparams, advparams, data=payload)
|
||||
current.id = partid
|
||||
return current
|
||||
if headersize:
|
||||
return self._readexact(headersize)
|
||||
return None
|
||||
|
||||
|
||||
class bundlepart(object):
|
||||
@ -584,6 +535,76 @@ class bundlepart(object):
|
||||
elif len(self.data):
|
||||
yield self.data
|
||||
|
||||
class unbundlepart(unpackermixin):
|
||||
"""a bundle part read from a bundle"""
|
||||
|
||||
def __init__(self, ui, header, fp):
|
||||
super(unbundlepart, self).__init__(fp)
|
||||
self.ui = ui
|
||||
# unbundle state attr
|
||||
self._headerdata = header
|
||||
# part data
|
||||
self.id = None
|
||||
self.type = None
|
||||
self.mandatoryparams = None
|
||||
self.advisoryparams = None
|
||||
self.data = None
|
||||
self._readdata()
|
||||
|
||||
def _readdata(self):
|
||||
"""read the header and setup the object"""
|
||||
# some utility to help reading from the header block
|
||||
headerblock = self._headerdata
|
||||
self._offset = 0 # layer violation to have something easy to understand
|
||||
def fromheader(size):
|
||||
"""return the next <size> byte from the header"""
|
||||
offset = self._offset
|
||||
data = headerblock[offset:(offset + size)]
|
||||
self._offset = offset + size
|
||||
return data
|
||||
def unpackheader(format):
|
||||
"""read given format from header
|
||||
|
||||
This automatically compute the size of the format to read."""
|
||||
data = fromheader(struct.calcsize(format))
|
||||
return _unpack(format, data)
|
||||
|
||||
typesize = unpackheader(_fparttypesize)[0]
|
||||
self.type = fromheader(typesize)
|
||||
self.ui.debug('part type: "%s"\n' % self.type)
|
||||
self.id = unpackheader(_fpartid)[0]
|
||||
self.ui.debug('part id: "%s"\n' % self.id)
|
||||
## reading parameters
|
||||
# param count
|
||||
mancount, advcount = unpackheader(_fpartparamcount)
|
||||
self.ui.debug('part parameters: %i\n' % (mancount + advcount))
|
||||
# param size
|
||||
paramsizes = unpackheader(_makefpartparamsizes(mancount + advcount))
|
||||
# make it a list of couple again
|
||||
paramsizes = zip(paramsizes[::2], paramsizes[1::2])
|
||||
# split mandatory from advisory
|
||||
mansizes = paramsizes[:mancount]
|
||||
advsizes = paramsizes[mancount:]
|
||||
# retrive param value
|
||||
manparams = []
|
||||
for key, value in mansizes:
|
||||
manparams.append((fromheader(key), fromheader(value)))
|
||||
advparams = []
|
||||
for key, value in advsizes:
|
||||
advparams.append((fromheader(key), fromheader(value)))
|
||||
del self._offset # clean up layer, nobody saw anything.
|
||||
self.mandatoryparams = manparams
|
||||
self.advisoryparams = advparams
|
||||
## part payload
|
||||
payload = []
|
||||
payloadsize = self._unpack(_fpayloadsize)[0]
|
||||
self.ui.debug('payload chunk size: %i\n' % payloadsize)
|
||||
while payloadsize:
|
||||
payload.append(self._readexact(payloadsize))
|
||||
payloadsize = self._unpack(_fpayloadsize)[0]
|
||||
self.ui.debug('payload chunk size: %i\n' % payloadsize)
|
||||
self.data = ''.join(payload)
|
||||
|
||||
@parthandler('changegroup')
|
||||
def handlechangegroup(op, inpart):
|
||||
"""apply a changegroup part on the repo
|
||||
@ -605,9 +626,10 @@ def handlechangegroup(op, inpart):
|
||||
if op.reply is not None:
|
||||
# This is definitly not the final form of this
|
||||
# return. But one need to start somewhere.
|
||||
op.reply.addpart(bundlepart('reply:changegroup', (),
|
||||
[('in-reply-to', str(inpart.id)),
|
||||
('return', '%i' % ret)]))
|
||||
part = bundlepart('reply:changegroup', (),
|
||||
[('in-reply-to', str(inpart.id)),
|
||||
('return', '%i' % ret)])
|
||||
op.reply.addpart(part)
|
||||
|
||||
@parthandler('reply:changegroup')
|
||||
def handlechangegroup(op, inpart):
|
||||
|
||||
Loading…
Reference in New Issue
Block a user