##// END OF EJS Templates
bundle2: make it possible to declare params handled by a part handler...
Pierre-Yves David -
r21623:5b26d82e default
parent child Browse files
Show More
@@ -1,845 +1,846
1 # bundle2.py - generic container format to transmit arbitrary data.
1 # bundle2.py - generic container format to transmit arbitrary data.
2 #
2 #
3 # Copyright 2013 Facebook, Inc.
3 # Copyright 2013 Facebook, Inc.
4 #
4 #
5 # This software may be used and distributed according to the terms of the
5 # This software may be used and distributed according to the terms of the
6 # GNU General Public License version 2 or any later version.
6 # GNU General Public License version 2 or any later version.
7 """Handling of the new bundle2 format
7 """Handling of the new bundle2 format
8
8
9 The goal of bundle2 is to act as an atomically packet to transmit a set of
9 The goal of bundle2 is to act as an atomically packet to transmit a set of
10 payloads in an application agnostic way. It consist in a sequence of "parts"
10 payloads in an application agnostic way. It consist in a sequence of "parts"
11 that will be handed to and processed by the application layer.
11 that will be handed to and processed by the application layer.
12
12
13
13
14 General format architecture
14 General format architecture
15 ===========================
15 ===========================
16
16
17 The format is architectured as follow
17 The format is architectured as follow
18
18
19 - magic string
19 - magic string
20 - stream level parameters
20 - stream level parameters
21 - payload parts (any number)
21 - payload parts (any number)
22 - end of stream marker.
22 - end of stream marker.
23
23
24 the Binary format
24 the Binary format
25 ============================
25 ============================
26
26
27 All numbers are unsigned and big-endian.
27 All numbers are unsigned and big-endian.
28
28
29 stream level parameters
29 stream level parameters
30 ------------------------
30 ------------------------
31
31
32 Binary format is as follow
32 Binary format is as follow
33
33
34 :params size: (16 bits integer)
34 :params size: (16 bits integer)
35
35
36 The total number of Bytes used by the parameters
36 The total number of Bytes used by the parameters
37
37
38 :params value: arbitrary number of Bytes
38 :params value: arbitrary number of Bytes
39
39
40 A blob of `params size` containing the serialized version of all stream level
40 A blob of `params size` containing the serialized version of all stream level
41 parameters.
41 parameters.
42
42
43 The blob contains a space separated list of parameters. Parameters with value
43 The blob contains a space separated list of parameters. Parameters with value
44 are stored in the form `<name>=<value>`. Both name and value are urlquoted.
44 are stored in the form `<name>=<value>`. Both name and value are urlquoted.
45
45
46 Empty name are obviously forbidden.
46 Empty name are obviously forbidden.
47
47
48 Name MUST start with a letter. If this first letter is lower case, the
48 Name MUST start with a letter. If this first letter is lower case, the
49 parameter is advisory and can be safely ignored. However when the first
49 parameter is advisory and can be safely ignored. However when the first
50 letter is capital, the parameter is mandatory and the bundling process MUST
50 letter is capital, the parameter is mandatory and the bundling process MUST
51 stop if he is not able to proceed it.
51 stop if he is not able to proceed it.
52
52
53 Stream parameters use a simple textual format for two main reasons:
53 Stream parameters use a simple textual format for two main reasons:
54
54
55 - Stream level parameters should remain simple and we want to discourage any
55 - Stream level parameters should remain simple and we want to discourage any
56 crazy usage.
56 crazy usage.
57 - Textual data allow easy human inspection of a bundle2 header in case of
57 - Textual data allow easy human inspection of a bundle2 header in case of
58 troubles.
58 troubles.
59
59
60 Any Applicative level options MUST go into a bundle2 part instead.
60 Any Applicative level options MUST go into a bundle2 part instead.
61
61
62 Payload part
62 Payload part
63 ------------------------
63 ------------------------
64
64
65 Binary format is as follow
65 Binary format is as follow
66
66
67 :header size: (16 bits inter)
67 :header size: (16 bits inter)
68
68
69 The total number of Bytes used by the part headers. When the header is empty
69 The total number of Bytes used by the part headers. When the header is empty
70 (size = 0) this is interpreted as the end of stream marker.
70 (size = 0) this is interpreted as the end of stream marker.
71
71
72 :header:
72 :header:
73
73
74 The header defines how to interpret the part. It contains two piece of
74 The header defines how to interpret the part. It contains two piece of
75 data: the part type, and the part parameters.
75 data: the part type, and the part parameters.
76
76
77 The part type is used to route an application level handler, that can
77 The part type is used to route an application level handler, that can
78 interpret payload.
78 interpret payload.
79
79
80 Part parameters are passed to the application level handler. They are
80 Part parameters are passed to the application level handler. They are
81 meant to convey information that will help the application level object to
81 meant to convey information that will help the application level object to
82 interpret the part payload.
82 interpret the part payload.
83
83
84 The binary format of the header is has follow
84 The binary format of the header is has follow
85
85
86 :typesize: (one byte)
86 :typesize: (one byte)
87
87
88 :parttype: alphanumerical part name
88 :parttype: alphanumerical part name
89
89
90 :partid: A 32bits integer (unique in the bundle) that can be used to refer
90 :partid: A 32bits integer (unique in the bundle) that can be used to refer
91 to this part.
91 to this part.
92
92
93 :parameters:
93 :parameters:
94
94
95 Part's parameter may have arbitrary content, the binary structure is::
95 Part's parameter may have arbitrary content, the binary structure is::
96
96
97 <mandatory-count><advisory-count><param-sizes><param-data>
97 <mandatory-count><advisory-count><param-sizes><param-data>
98
98
99 :mandatory-count: 1 byte, number of mandatory parameters
99 :mandatory-count: 1 byte, number of mandatory parameters
100
100
101 :advisory-count: 1 byte, number of advisory parameters
101 :advisory-count: 1 byte, number of advisory parameters
102
102
103 :param-sizes:
103 :param-sizes:
104
104
105 N couple of bytes, where N is the total number of parameters. Each
105 N couple of bytes, where N is the total number of parameters. Each
106 couple contains (<size-of-key>, <size-of-value) for one parameter.
106 couple contains (<size-of-key>, <size-of-value) for one parameter.
107
107
108 :param-data:
108 :param-data:
109
109
110 A blob of bytes from which each parameter key and value can be
110 A blob of bytes from which each parameter key and value can be
111 retrieved using the list of size couples stored in the previous
111 retrieved using the list of size couples stored in the previous
112 field.
112 field.
113
113
114 Mandatory parameters comes first, then the advisory ones.
114 Mandatory parameters comes first, then the advisory ones.
115
115
116 Each parameter's key MUST be unique within the part.
116 Each parameter's key MUST be unique within the part.
117
117
118 :payload:
118 :payload:
119
119
120 payload is a series of `<chunksize><chunkdata>`.
120 payload is a series of `<chunksize><chunkdata>`.
121
121
122 `chunksize` is a 32 bits integer, `chunkdata` are plain bytes (as much as
122 `chunksize` is a 32 bits integer, `chunkdata` are plain bytes (as much as
123 `chunksize` says)` The payload part is concluded by a zero size chunk.
123 `chunksize` says)` The payload part is concluded by a zero size chunk.
124
124
125 The current implementation always produces either zero or one chunk.
125 The current implementation always produces either zero or one chunk.
126 This is an implementation limitation that will ultimately be lifted.
126 This is an implementation limitation that will ultimately be lifted.
127
127
128 Bundle processing
128 Bundle processing
129 ============================
129 ============================
130
130
131 Each part is processed in order using a "part handler". Handler are registered
131 Each part is processed in order using a "part handler". Handler are registered
132 for a certain part type.
132 for a certain part type.
133
133
134 The matching of a part to its handler is case insensitive. The case of the
134 The matching of a part to its handler is case insensitive. The case of the
135 part type is used to know if a part is mandatory or advisory. If the Part type
135 part type is used to know if a part is mandatory or advisory. If the Part type
136 contains any uppercase char it is considered mandatory. When no handler is
136 contains any uppercase char it is considered mandatory. When no handler is
137 known for a Mandatory part, the process is aborted and an exception is raised.
137 known for a Mandatory part, the process is aborted and an exception is raised.
138 If the part is advisory and no handler is known, the part is ignored. When the
138 If the part is advisory and no handler is known, the part is ignored. When the
139 process is aborted, the full bundle is still read from the stream to keep the
139 process is aborted, the full bundle is still read from the stream to keep the
140 channel usable. But none of the part read from an abort are processed. In the
140 channel usable. But none of the part read from an abort are processed. In the
141 future, dropping the stream may become an option for channel we do not care to
141 future, dropping the stream may become an option for channel we do not care to
142 preserve.
142 preserve.
143 """
143 """
144
144
145 import util
145 import util
146 import struct
146 import struct
147 import urllib
147 import urllib
148 import string
148 import string
149
149
150 import changegroup, error
150 import changegroup, error
151 from i18n import _
151 from i18n import _
152
152
153 _pack = struct.pack
153 _pack = struct.pack
154 _unpack = struct.unpack
154 _unpack = struct.unpack
155
155
156 _magicstring = 'HG2X'
156 _magicstring = 'HG2X'
157
157
158 _fstreamparamsize = '>H'
158 _fstreamparamsize = '>H'
159 _fpartheadersize = '>H'
159 _fpartheadersize = '>H'
160 _fparttypesize = '>B'
160 _fparttypesize = '>B'
161 _fpartid = '>I'
161 _fpartid = '>I'
162 _fpayloadsize = '>I'
162 _fpayloadsize = '>I'
163 _fpartparamcount = '>BB'
163 _fpartparamcount = '>BB'
164
164
165 preferedchunksize = 4096
165 preferedchunksize = 4096
166
166
167 def _makefpartparamsizes(nbparams):
167 def _makefpartparamsizes(nbparams):
168 """return a struct format to read part parameter sizes
168 """return a struct format to read part parameter sizes
169
169
170 The number parameters is variable so we need to build that format
170 The number parameters is variable so we need to build that format
171 dynamically.
171 dynamically.
172 """
172 """
173 return '>'+('BB'*nbparams)
173 return '>'+('BB'*nbparams)
174
174
175 parthandlermapping = {}
175 parthandlermapping = {}
176
176
177 def parthandler(parttype):
177 def parthandler(parttype, params=()):
178 """decorator that register a function as a bundle2 part handler
178 """decorator that register a function as a bundle2 part handler
179
179
180 eg::
180 eg::
181
181
182 @parthandler('myparttype')
182 @parthandler('myparttype')
183 def myparttypehandler(...):
183 def myparttypehandler(...):
184 '''process a part of type "my part".'''
184 '''process a part of type "my part".'''
185 ...
185 ...
186 """
186 """
187 def _decorator(func):
187 def _decorator(func):
188 lparttype = parttype.lower() # enforce lower case matching.
188 lparttype = parttype.lower() # enforce lower case matching.
189 assert lparttype not in parthandlermapping
189 assert lparttype not in parthandlermapping
190 parthandlermapping[lparttype] = func
190 parthandlermapping[lparttype] = func
191 func.params = frozenset(params)
191 return func
192 return func
192 return _decorator
193 return _decorator
193
194
194 class unbundlerecords(object):
195 class unbundlerecords(object):
195 """keep record of what happens during and unbundle
196 """keep record of what happens during and unbundle
196
197
197 New records are added using `records.add('cat', obj)`. Where 'cat' is a
198 New records are added using `records.add('cat', obj)`. Where 'cat' is a
198 category of record and obj is an arbitrary object.
199 category of record and obj is an arbitrary object.
199
200
200 `records['cat']` will return all entries of this category 'cat'.
201 `records['cat']` will return all entries of this category 'cat'.
201
202
202 Iterating on the object itself will yield `('category', obj)` tuples
203 Iterating on the object itself will yield `('category', obj)` tuples
203 for all entries.
204 for all entries.
204
205
205 All iterations happens in chronological order.
206 All iterations happens in chronological order.
206 """
207 """
207
208
208 def __init__(self):
209 def __init__(self):
209 self._categories = {}
210 self._categories = {}
210 self._sequences = []
211 self._sequences = []
211 self._replies = {}
212 self._replies = {}
212
213
213 def add(self, category, entry, inreplyto=None):
214 def add(self, category, entry, inreplyto=None):
214 """add a new record of a given category.
215 """add a new record of a given category.
215
216
216 The entry can then be retrieved in the list returned by
217 The entry can then be retrieved in the list returned by
217 self['category']."""
218 self['category']."""
218 self._categories.setdefault(category, []).append(entry)
219 self._categories.setdefault(category, []).append(entry)
219 self._sequences.append((category, entry))
220 self._sequences.append((category, entry))
220 if inreplyto is not None:
221 if inreplyto is not None:
221 self.getreplies(inreplyto).add(category, entry)
222 self.getreplies(inreplyto).add(category, entry)
222
223
223 def getreplies(self, partid):
224 def getreplies(self, partid):
224 """get the subrecords that replies to a specific part"""
225 """get the subrecords that replies to a specific part"""
225 return self._replies.setdefault(partid, unbundlerecords())
226 return self._replies.setdefault(partid, unbundlerecords())
226
227
227 def __getitem__(self, cat):
228 def __getitem__(self, cat):
228 return tuple(self._categories.get(cat, ()))
229 return tuple(self._categories.get(cat, ()))
229
230
230 def __iter__(self):
231 def __iter__(self):
231 return iter(self._sequences)
232 return iter(self._sequences)
232
233
233 def __len__(self):
234 def __len__(self):
234 return len(self._sequences)
235 return len(self._sequences)
235
236
236 def __nonzero__(self):
237 def __nonzero__(self):
237 return bool(self._sequences)
238 return bool(self._sequences)
238
239
239 class bundleoperation(object):
240 class bundleoperation(object):
240 """an object that represents a single bundling process
241 """an object that represents a single bundling process
241
242
242 Its purpose is to carry unbundle-related objects and states.
243 Its purpose is to carry unbundle-related objects and states.
243
244
244 A new object should be created at the beginning of each bundle processing.
245 A new object should be created at the beginning of each bundle processing.
245 The object is to be returned by the processing function.
246 The object is to be returned by the processing function.
246
247
247 The object has very little content now it will ultimately contain:
248 The object has very little content now it will ultimately contain:
248 * an access to the repo the bundle is applied to,
249 * an access to the repo the bundle is applied to,
249 * a ui object,
250 * a ui object,
250 * a way to retrieve a transaction to add changes to the repo,
251 * a way to retrieve a transaction to add changes to the repo,
251 * a way to record the result of processing each part,
252 * a way to record the result of processing each part,
252 * a way to construct a bundle response when applicable.
253 * a way to construct a bundle response when applicable.
253 """
254 """
254
255
255 def __init__(self, repo, transactiongetter):
256 def __init__(self, repo, transactiongetter):
256 self.repo = repo
257 self.repo = repo
257 self.ui = repo.ui
258 self.ui = repo.ui
258 self.records = unbundlerecords()
259 self.records = unbundlerecords()
259 self.gettransaction = transactiongetter
260 self.gettransaction = transactiongetter
260 self.reply = None
261 self.reply = None
261
262
262 class TransactionUnavailable(RuntimeError):
263 class TransactionUnavailable(RuntimeError):
263 pass
264 pass
264
265
265 def _notransaction():
266 def _notransaction():
266 """default method to get a transaction while processing a bundle
267 """default method to get a transaction while processing a bundle
267
268
268 Raise an exception to highlight the fact that no transaction was expected
269 Raise an exception to highlight the fact that no transaction was expected
269 to be created"""
270 to be created"""
270 raise TransactionUnavailable()
271 raise TransactionUnavailable()
271
272
272 def processbundle(repo, unbundler, transactiongetter=_notransaction):
273 def processbundle(repo, unbundler, transactiongetter=_notransaction):
273 """This function process a bundle, apply effect to/from a repo
274 """This function process a bundle, apply effect to/from a repo
274
275
275 It iterates over each part then searches for and uses the proper handling
276 It iterates over each part then searches for and uses the proper handling
276 code to process the part. Parts are processed in order.
277 code to process the part. Parts are processed in order.
277
278
278 This is very early version of this function that will be strongly reworked
279 This is very early version of this function that will be strongly reworked
279 before final usage.
280 before final usage.
280
281
281 Unknown Mandatory part will abort the process.
282 Unknown Mandatory part will abort the process.
282 """
283 """
283 op = bundleoperation(repo, transactiongetter)
284 op = bundleoperation(repo, transactiongetter)
284 # todo:
285 # todo:
285 # - replace this is a init function soon.
286 # - replace this is a init function soon.
286 # - exception catching
287 # - exception catching
287 unbundler.params
288 unbundler.params
288 iterparts = unbundler.iterparts()
289 iterparts = unbundler.iterparts()
289 part = None
290 part = None
290 try:
291 try:
291 for part in iterparts:
292 for part in iterparts:
292 parttype = part.type
293 parttype = part.type
293 # part key are matched lower case
294 # part key are matched lower case
294 key = parttype.lower()
295 key = parttype.lower()
295 try:
296 try:
296 handler = parthandlermapping[key]
297 handler = parthandlermapping[key]
297 op.ui.debug('found a handler for part %r\n' % parttype)
298 op.ui.debug('found a handler for part %r\n' % parttype)
298 except KeyError:
299 except KeyError:
299 if key != parttype: # mandatory parts
300 if key != parttype: # mandatory parts
300 # todo:
301 # todo:
301 # - use a more precise exception
302 # - use a more precise exception
302 raise error.BundleValueError(parttype=key)
303 raise error.BundleValueError(parttype=key)
303 op.ui.debug('ignoring unknown advisory part %r\n' % key)
304 op.ui.debug('ignoring unknown advisory part %r\n' % key)
304 # consuming the part
305 # consuming the part
305 part.read()
306 part.read()
306 continue
307 continue
307
308
308 # handler is called outside the above try block so that we don't
309 # handler is called outside the above try block so that we don't
309 # risk catching KeyErrors from anything other than the
310 # risk catching KeyErrors from anything other than the
310 # parthandlermapping lookup (any KeyError raised by handler()
311 # parthandlermapping lookup (any KeyError raised by handler()
311 # itself represents a defect of a different variety).
312 # itself represents a defect of a different variety).
312 output = None
313 output = None
313 if op.reply is not None:
314 if op.reply is not None:
314 op.ui.pushbuffer(error=True)
315 op.ui.pushbuffer(error=True)
315 output = ''
316 output = ''
316 try:
317 try:
317 handler(op, part)
318 handler(op, part)
318 finally:
319 finally:
319 if output is not None:
320 if output is not None:
320 output = op.ui.popbuffer()
321 output = op.ui.popbuffer()
321 if output:
322 if output:
322 outpart = op.reply.newpart('b2x:output', data=output)
323 outpart = op.reply.newpart('b2x:output', data=output)
323 outpart.addparam('in-reply-to', str(part.id), mandatory=False)
324 outpart.addparam('in-reply-to', str(part.id), mandatory=False)
324 part.read()
325 part.read()
325 except Exception, exc:
326 except Exception, exc:
326 if part is not None:
327 if part is not None:
327 # consume the bundle content
328 # consume the bundle content
328 part.read()
329 part.read()
329 for part in iterparts:
330 for part in iterparts:
330 # consume the bundle content
331 # consume the bundle content
331 part.read()
332 part.read()
332 # Small hack to let caller code distinguish exceptions from bundle2
333 # Small hack to let caller code distinguish exceptions from bundle2
333 # processing fron the ones from bundle1 processing. This is mostly
334 # processing fron the ones from bundle1 processing. This is mostly
334 # needed to handle different return codes to unbundle according to the
335 # needed to handle different return codes to unbundle according to the
335 # type of bundle. We should probably clean up or drop this return code
336 # type of bundle. We should probably clean up or drop this return code
336 # craziness in a future version.
337 # craziness in a future version.
337 exc.duringunbundle2 = True
338 exc.duringunbundle2 = True
338 raise
339 raise
339 return op
340 return op
340
341
341 def decodecaps(blob):
342 def decodecaps(blob):
342 """decode a bundle2 caps bytes blob into a dictionnary
343 """decode a bundle2 caps bytes blob into a dictionnary
343
344
344 The blob is a list of capabilities (one per line)
345 The blob is a list of capabilities (one per line)
345 Capabilities may have values using a line of the form::
346 Capabilities may have values using a line of the form::
346
347
347 capability=value1,value2,value3
348 capability=value1,value2,value3
348
349
349 The values are always a list."""
350 The values are always a list."""
350 caps = {}
351 caps = {}
351 for line in blob.splitlines():
352 for line in blob.splitlines():
352 if not line:
353 if not line:
353 continue
354 continue
354 if '=' not in line:
355 if '=' not in line:
355 key, vals = line, ()
356 key, vals = line, ()
356 else:
357 else:
357 key, vals = line.split('=', 1)
358 key, vals = line.split('=', 1)
358 vals = vals.split(',')
359 vals = vals.split(',')
359 key = urllib.unquote(key)
360 key = urllib.unquote(key)
360 vals = [urllib.unquote(v) for v in vals]
361 vals = [urllib.unquote(v) for v in vals]
361 caps[key] = vals
362 caps[key] = vals
362 return caps
363 return caps
363
364
364 def encodecaps(caps):
365 def encodecaps(caps):
365 """encode a bundle2 caps dictionary into a bytes blob"""
366 """encode a bundle2 caps dictionary into a bytes blob"""
366 chunks = []
367 chunks = []
367 for ca in sorted(caps):
368 for ca in sorted(caps):
368 vals = caps[ca]
369 vals = caps[ca]
369 ca = urllib.quote(ca)
370 ca = urllib.quote(ca)
370 vals = [urllib.quote(v) for v in vals]
371 vals = [urllib.quote(v) for v in vals]
371 if vals:
372 if vals:
372 ca = "%s=%s" % (ca, ','.join(vals))
373 ca = "%s=%s" % (ca, ','.join(vals))
373 chunks.append(ca)
374 chunks.append(ca)
374 return '\n'.join(chunks)
375 return '\n'.join(chunks)
375
376
376 class bundle20(object):
377 class bundle20(object):
377 """represent an outgoing bundle2 container
378 """represent an outgoing bundle2 container
378
379
379 Use the `addparam` method to add stream level parameter. and `newpart` to
380 Use the `addparam` method to add stream level parameter. and `newpart` to
380 populate it. Then call `getchunks` to retrieve all the binary chunks of
381 populate it. Then call `getchunks` to retrieve all the binary chunks of
381 data that compose the bundle2 container."""
382 data that compose the bundle2 container."""
382
383
383 def __init__(self, ui, capabilities=()):
384 def __init__(self, ui, capabilities=()):
384 self.ui = ui
385 self.ui = ui
385 self._params = []
386 self._params = []
386 self._parts = []
387 self._parts = []
387 self.capabilities = dict(capabilities)
388 self.capabilities = dict(capabilities)
388
389
389 # methods used to defines the bundle2 content
390 # methods used to defines the bundle2 content
390 def addparam(self, name, value=None):
391 def addparam(self, name, value=None):
391 """add a stream level parameter"""
392 """add a stream level parameter"""
392 if not name:
393 if not name:
393 raise ValueError('empty parameter name')
394 raise ValueError('empty parameter name')
394 if name[0] not in string.letters:
395 if name[0] not in string.letters:
395 raise ValueError('non letter first character: %r' % name)
396 raise ValueError('non letter first character: %r' % name)
396 self._params.append((name, value))
397 self._params.append((name, value))
397
398
398 def addpart(self, part):
399 def addpart(self, part):
399 """add a new part to the bundle2 container
400 """add a new part to the bundle2 container
400
401
401 Parts contains the actual applicative payload."""
402 Parts contains the actual applicative payload."""
402 assert part.id is None
403 assert part.id is None
403 part.id = len(self._parts) # very cheap counter
404 part.id = len(self._parts) # very cheap counter
404 self._parts.append(part)
405 self._parts.append(part)
405
406
406 def newpart(self, typeid, *args, **kwargs):
407 def newpart(self, typeid, *args, **kwargs):
407 """create a new part and add it to the containers
408 """create a new part and add it to the containers
408
409
409 As the part is directly added to the containers. For now, this means
410 As the part is directly added to the containers. For now, this means
410 that any failure to properly initialize the part after calling
411 that any failure to properly initialize the part after calling
411 ``newpart`` should result in a failure of the whole bundling process.
412 ``newpart`` should result in a failure of the whole bundling process.
412
413
413 You can still fall back to manually create and add if you need better
414 You can still fall back to manually create and add if you need better
414 control."""
415 control."""
415 part = bundlepart(typeid, *args, **kwargs)
416 part = bundlepart(typeid, *args, **kwargs)
416 self.addpart(part)
417 self.addpart(part)
417 return part
418 return part
418
419
419 # methods used to generate the bundle2 stream
420 # methods used to generate the bundle2 stream
420 def getchunks(self):
421 def getchunks(self):
421 self.ui.debug('start emission of %s stream\n' % _magicstring)
422 self.ui.debug('start emission of %s stream\n' % _magicstring)
422 yield _magicstring
423 yield _magicstring
423 param = self._paramchunk()
424 param = self._paramchunk()
424 self.ui.debug('bundle parameter: %s\n' % param)
425 self.ui.debug('bundle parameter: %s\n' % param)
425 yield _pack(_fstreamparamsize, len(param))
426 yield _pack(_fstreamparamsize, len(param))
426 if param:
427 if param:
427 yield param
428 yield param
428
429
429 self.ui.debug('start of parts\n')
430 self.ui.debug('start of parts\n')
430 for part in self._parts:
431 for part in self._parts:
431 self.ui.debug('bundle part: "%s"\n' % part.type)
432 self.ui.debug('bundle part: "%s"\n' % part.type)
432 for chunk in part.getchunks():
433 for chunk in part.getchunks():
433 yield chunk
434 yield chunk
434 self.ui.debug('end of bundle\n')
435 self.ui.debug('end of bundle\n')
435 yield '\0\0'
436 yield '\0\0'
436
437
437 def _paramchunk(self):
438 def _paramchunk(self):
438 """return a encoded version of all stream parameters"""
439 """return a encoded version of all stream parameters"""
439 blocks = []
440 blocks = []
440 for par, value in self._params:
441 for par, value in self._params:
441 par = urllib.quote(par)
442 par = urllib.quote(par)
442 if value is not None:
443 if value is not None:
443 value = urllib.quote(value)
444 value = urllib.quote(value)
444 par = '%s=%s' % (par, value)
445 par = '%s=%s' % (par, value)
445 blocks.append(par)
446 blocks.append(par)
446 return ' '.join(blocks)
447 return ' '.join(blocks)
447
448
448 class unpackermixin(object):
449 class unpackermixin(object):
449 """A mixin to extract bytes and struct data from a stream"""
450 """A mixin to extract bytes and struct data from a stream"""
450
451
451 def __init__(self, fp):
452 def __init__(self, fp):
452 self._fp = fp
453 self._fp = fp
453
454
454 def _unpack(self, format):
455 def _unpack(self, format):
455 """unpack this struct format from the stream"""
456 """unpack this struct format from the stream"""
456 data = self._readexact(struct.calcsize(format))
457 data = self._readexact(struct.calcsize(format))
457 return _unpack(format, data)
458 return _unpack(format, data)
458
459
459 def _readexact(self, size):
460 def _readexact(self, size):
460 """read exactly <size> bytes from the stream"""
461 """read exactly <size> bytes from the stream"""
461 return changegroup.readexactly(self._fp, size)
462 return changegroup.readexactly(self._fp, size)
462
463
463
464
464 class unbundle20(unpackermixin):
465 class unbundle20(unpackermixin):
465 """interpret a bundle2 stream
466 """interpret a bundle2 stream
466
467
467 This class is fed with a binary stream and yields parts through its
468 This class is fed with a binary stream and yields parts through its
468 `iterparts` methods."""
469 `iterparts` methods."""
469
470
470 def __init__(self, ui, fp, header=None):
471 def __init__(self, ui, fp, header=None):
471 """If header is specified, we do not read it out of the stream."""
472 """If header is specified, we do not read it out of the stream."""
472 self.ui = ui
473 self.ui = ui
473 super(unbundle20, self).__init__(fp)
474 super(unbundle20, self).__init__(fp)
474 if header is None:
475 if header is None:
475 header = self._readexact(4)
476 header = self._readexact(4)
476 magic, version = header[0:2], header[2:4]
477 magic, version = header[0:2], header[2:4]
477 if magic != 'HG':
478 if magic != 'HG':
478 raise util.Abort(_('not a Mercurial bundle'))
479 raise util.Abort(_('not a Mercurial bundle'))
479 if version != '2X':
480 if version != '2X':
480 raise util.Abort(_('unknown bundle version %s') % version)
481 raise util.Abort(_('unknown bundle version %s') % version)
481 self.ui.debug('start processing of %s stream\n' % header)
482 self.ui.debug('start processing of %s stream\n' % header)
482
483
483 @util.propertycache
484 @util.propertycache
484 def params(self):
485 def params(self):
485 """dictionary of stream level parameters"""
486 """dictionary of stream level parameters"""
486 self.ui.debug('reading bundle2 stream parameters\n')
487 self.ui.debug('reading bundle2 stream parameters\n')
487 params = {}
488 params = {}
488 paramssize = self._unpack(_fstreamparamsize)[0]
489 paramssize = self._unpack(_fstreamparamsize)[0]
489 if paramssize:
490 if paramssize:
490 for p in self._readexact(paramssize).split(' '):
491 for p in self._readexact(paramssize).split(' '):
491 p = p.split('=', 1)
492 p = p.split('=', 1)
492 p = [urllib.unquote(i) for i in p]
493 p = [urllib.unquote(i) for i in p]
493 if len(p) < 2:
494 if len(p) < 2:
494 p.append(None)
495 p.append(None)
495 self._processparam(*p)
496 self._processparam(*p)
496 params[p[0]] = p[1]
497 params[p[0]] = p[1]
497 return params
498 return params
498
499
499 def _processparam(self, name, value):
500 def _processparam(self, name, value):
500 """process a parameter, applying its effect if needed
501 """process a parameter, applying its effect if needed
501
502
502 Parameter starting with a lower case letter are advisory and will be
503 Parameter starting with a lower case letter are advisory and will be
503 ignored when unknown. Those starting with an upper case letter are
504 ignored when unknown. Those starting with an upper case letter are
504 mandatory and will this function will raise a KeyError when unknown.
505 mandatory and will this function will raise a KeyError when unknown.
505
506
506 Note: no option are currently supported. Any input will be either
507 Note: no option are currently supported. Any input will be either
507 ignored or failing.
508 ignored or failing.
508 """
509 """
509 if not name:
510 if not name:
510 raise ValueError('empty parameter name')
511 raise ValueError('empty parameter name')
511 if name[0] not in string.letters:
512 if name[0] not in string.letters:
512 raise ValueError('non letter first character: %r' % name)
513 raise ValueError('non letter first character: %r' % name)
513 # Some logic will be later added here to try to process the option for
514 # Some logic will be later added here to try to process the option for
514 # a dict of known parameter.
515 # a dict of known parameter.
515 if name[0].islower():
516 if name[0].islower():
516 self.ui.debug("ignoring unknown parameter %r\n" % name)
517 self.ui.debug("ignoring unknown parameter %r\n" % name)
517 else:
518 else:
518 raise KeyError(name)
519 raise KeyError(name)
519
520
520
521
521 def iterparts(self):
522 def iterparts(self):
522 """yield all parts contained in the stream"""
523 """yield all parts contained in the stream"""
523 # make sure param have been loaded
524 # make sure param have been loaded
524 self.params
525 self.params
525 self.ui.debug('start extraction of bundle2 parts\n')
526 self.ui.debug('start extraction of bundle2 parts\n')
526 headerblock = self._readpartheader()
527 headerblock = self._readpartheader()
527 while headerblock is not None:
528 while headerblock is not None:
528 part = unbundlepart(self.ui, headerblock, self._fp)
529 part = unbundlepart(self.ui, headerblock, self._fp)
529 yield part
530 yield part
530 headerblock = self._readpartheader()
531 headerblock = self._readpartheader()
531 self.ui.debug('end of bundle2 stream\n')
532 self.ui.debug('end of bundle2 stream\n')
532
533
533 def _readpartheader(self):
534 def _readpartheader(self):
534 """reads a part header size and return the bytes blob
535 """reads a part header size and return the bytes blob
535
536
536 returns None if empty"""
537 returns None if empty"""
537 headersize = self._unpack(_fpartheadersize)[0]
538 headersize = self._unpack(_fpartheadersize)[0]
538 self.ui.debug('part header size: %i\n' % headersize)
539 self.ui.debug('part header size: %i\n' % headersize)
539 if headersize:
540 if headersize:
540 return self._readexact(headersize)
541 return self._readexact(headersize)
541 return None
542 return None
542
543
543
544
544 class bundlepart(object):
545 class bundlepart(object):
545 """A bundle2 part contains application level payload
546 """A bundle2 part contains application level payload
546
547
547 The part `type` is used to route the part to the application level
548 The part `type` is used to route the part to the application level
548 handler.
549 handler.
549
550
550 The part payload is contained in ``part.data``. It could be raw bytes or a
551 The part payload is contained in ``part.data``. It could be raw bytes or a
551 generator of byte chunks.
552 generator of byte chunks.
552
553
553 You can add parameters to the part using the ``addparam`` method.
554 You can add parameters to the part using the ``addparam`` method.
554 Parameters can be either mandatory (default) or advisory. Remote side
555 Parameters can be either mandatory (default) or advisory. Remote side
555 should be able to safely ignore the advisory ones.
556 should be able to safely ignore the advisory ones.
556
557
557 Both data and parameters cannot be modified after the generation has begun.
558 Both data and parameters cannot be modified after the generation has begun.
558 """
559 """
559
560
560 def __init__(self, parttype, mandatoryparams=(), advisoryparams=(),
561 def __init__(self, parttype, mandatoryparams=(), advisoryparams=(),
561 data=''):
562 data=''):
562 self.id = None
563 self.id = None
563 self.type = parttype
564 self.type = parttype
564 self._data = data
565 self._data = data
565 self._mandatoryparams = list(mandatoryparams)
566 self._mandatoryparams = list(mandatoryparams)
566 self._advisoryparams = list(advisoryparams)
567 self._advisoryparams = list(advisoryparams)
567 # checking for duplicated entries
568 # checking for duplicated entries
568 self._seenparams = set()
569 self._seenparams = set()
569 for pname, __ in self._mandatoryparams + self._advisoryparams:
570 for pname, __ in self._mandatoryparams + self._advisoryparams:
570 if pname in self._seenparams:
571 if pname in self._seenparams:
571 raise RuntimeError('duplicated params: %s' % pname)
572 raise RuntimeError('duplicated params: %s' % pname)
572 self._seenparams.add(pname)
573 self._seenparams.add(pname)
573 # status of the part's generation:
574 # status of the part's generation:
574 # - None: not started,
575 # - None: not started,
575 # - False: currently generated,
576 # - False: currently generated,
576 # - True: generation done.
577 # - True: generation done.
577 self._generated = None
578 self._generated = None
578
579
579 # methods used to defines the part content
580 # methods used to defines the part content
580 def __setdata(self, data):
581 def __setdata(self, data):
581 if self._generated is not None:
582 if self._generated is not None:
582 raise error.ReadOnlyPartError('part is being generated')
583 raise error.ReadOnlyPartError('part is being generated')
583 self._data = data
584 self._data = data
584 def __getdata(self):
585 def __getdata(self):
585 return self._data
586 return self._data
586 data = property(__getdata, __setdata)
587 data = property(__getdata, __setdata)
587
588
588 @property
589 @property
589 def mandatoryparams(self):
590 def mandatoryparams(self):
590 # make it an immutable tuple to force people through ``addparam``
591 # make it an immutable tuple to force people through ``addparam``
591 return tuple(self._mandatoryparams)
592 return tuple(self._mandatoryparams)
592
593
593 @property
594 @property
594 def advisoryparams(self):
595 def advisoryparams(self):
595 # make it an immutable tuple to force people through ``addparam``
596 # make it an immutable tuple to force people through ``addparam``
596 return tuple(self._advisoryparams)
597 return tuple(self._advisoryparams)
597
598
598 def addparam(self, name, value='', mandatory=True):
599 def addparam(self, name, value='', mandatory=True):
599 if self._generated is not None:
600 if self._generated is not None:
600 raise error.ReadOnlyPartError('part is being generated')
601 raise error.ReadOnlyPartError('part is being generated')
601 if name in self._seenparams:
602 if name in self._seenparams:
602 raise ValueError('duplicated params: %s' % name)
603 raise ValueError('duplicated params: %s' % name)
603 self._seenparams.add(name)
604 self._seenparams.add(name)
604 params = self._advisoryparams
605 params = self._advisoryparams
605 if mandatory:
606 if mandatory:
606 params = self._mandatoryparams
607 params = self._mandatoryparams
607 params.append((name, value))
608 params.append((name, value))
608
609
609 # methods used to generates the bundle2 stream
610 # methods used to generates the bundle2 stream
610 def getchunks(self):
611 def getchunks(self):
611 if self._generated is not None:
612 if self._generated is not None:
612 raise RuntimeError('part can only be consumed once')
613 raise RuntimeError('part can only be consumed once')
613 self._generated = False
614 self._generated = False
614 #### header
615 #### header
615 ## parttype
616 ## parttype
616 header = [_pack(_fparttypesize, len(self.type)),
617 header = [_pack(_fparttypesize, len(self.type)),
617 self.type, _pack(_fpartid, self.id),
618 self.type, _pack(_fpartid, self.id),
618 ]
619 ]
619 ## parameters
620 ## parameters
620 # count
621 # count
621 manpar = self.mandatoryparams
622 manpar = self.mandatoryparams
622 advpar = self.advisoryparams
623 advpar = self.advisoryparams
623 header.append(_pack(_fpartparamcount, len(manpar), len(advpar)))
624 header.append(_pack(_fpartparamcount, len(manpar), len(advpar)))
624 # size
625 # size
625 parsizes = []
626 parsizes = []
626 for key, value in manpar:
627 for key, value in manpar:
627 parsizes.append(len(key))
628 parsizes.append(len(key))
628 parsizes.append(len(value))
629 parsizes.append(len(value))
629 for key, value in advpar:
630 for key, value in advpar:
630 parsizes.append(len(key))
631 parsizes.append(len(key))
631 parsizes.append(len(value))
632 parsizes.append(len(value))
632 paramsizes = _pack(_makefpartparamsizes(len(parsizes) / 2), *parsizes)
633 paramsizes = _pack(_makefpartparamsizes(len(parsizes) / 2), *parsizes)
633 header.append(paramsizes)
634 header.append(paramsizes)
634 # key, value
635 # key, value
635 for key, value in manpar:
636 for key, value in manpar:
636 header.append(key)
637 header.append(key)
637 header.append(value)
638 header.append(value)
638 for key, value in advpar:
639 for key, value in advpar:
639 header.append(key)
640 header.append(key)
640 header.append(value)
641 header.append(value)
641 ## finalize header
642 ## finalize header
642 headerchunk = ''.join(header)
643 headerchunk = ''.join(header)
643 yield _pack(_fpartheadersize, len(headerchunk))
644 yield _pack(_fpartheadersize, len(headerchunk))
644 yield headerchunk
645 yield headerchunk
645 ## payload
646 ## payload
646 for chunk in self._payloadchunks():
647 for chunk in self._payloadchunks():
647 yield _pack(_fpayloadsize, len(chunk))
648 yield _pack(_fpayloadsize, len(chunk))
648 yield chunk
649 yield chunk
649 # end of payload
650 # end of payload
650 yield _pack(_fpayloadsize, 0)
651 yield _pack(_fpayloadsize, 0)
651 self._generated = True
652 self._generated = True
652
653
653 def _payloadchunks(self):
654 def _payloadchunks(self):
654 """yield chunks of a the part payload
655 """yield chunks of a the part payload
655
656
656 Exists to handle the different methods to provide data to a part."""
657 Exists to handle the different methods to provide data to a part."""
657 # we only support fixed size data now.
658 # we only support fixed size data now.
658 # This will be improved in the future.
659 # This will be improved in the future.
659 if util.safehasattr(self.data, 'next'):
660 if util.safehasattr(self.data, 'next'):
660 buff = util.chunkbuffer(self.data)
661 buff = util.chunkbuffer(self.data)
661 chunk = buff.read(preferedchunksize)
662 chunk = buff.read(preferedchunksize)
662 while chunk:
663 while chunk:
663 yield chunk
664 yield chunk
664 chunk = buff.read(preferedchunksize)
665 chunk = buff.read(preferedchunksize)
665 elif len(self.data):
666 elif len(self.data):
666 yield self.data
667 yield self.data
667
668
668 class unbundlepart(unpackermixin):
669 class unbundlepart(unpackermixin):
669 """a bundle part read from a bundle"""
670 """a bundle part read from a bundle"""
670
671
671 def __init__(self, ui, header, fp):
672 def __init__(self, ui, header, fp):
672 super(unbundlepart, self).__init__(fp)
673 super(unbundlepart, self).__init__(fp)
673 self.ui = ui
674 self.ui = ui
674 # unbundle state attr
675 # unbundle state attr
675 self._headerdata = header
676 self._headerdata = header
676 self._headeroffset = 0
677 self._headeroffset = 0
677 self._initialized = False
678 self._initialized = False
678 self.consumed = False
679 self.consumed = False
679 # part data
680 # part data
680 self.id = None
681 self.id = None
681 self.type = None
682 self.type = None
682 self.mandatoryparams = None
683 self.mandatoryparams = None
683 self.advisoryparams = None
684 self.advisoryparams = None
684 self.params = None
685 self.params = None
685 self.mandatorykeys = ()
686 self.mandatorykeys = ()
686 self._payloadstream = None
687 self._payloadstream = None
687 self._readheader()
688 self._readheader()
688
689
689 def _fromheader(self, size):
690 def _fromheader(self, size):
690 """return the next <size> byte from the header"""
691 """return the next <size> byte from the header"""
691 offset = self._headeroffset
692 offset = self._headeroffset
692 data = self._headerdata[offset:(offset + size)]
693 data = self._headerdata[offset:(offset + size)]
693 self._headeroffset = offset + size
694 self._headeroffset = offset + size
694 return data
695 return data
695
696
696 def _unpackheader(self, format):
697 def _unpackheader(self, format):
697 """read given format from header
698 """read given format from header
698
699
699 This automatically compute the size of the format to read."""
700 This automatically compute the size of the format to read."""
700 data = self._fromheader(struct.calcsize(format))
701 data = self._fromheader(struct.calcsize(format))
701 return _unpack(format, data)
702 return _unpack(format, data)
702
703
703 def _initparams(self, mandatoryparams, advisoryparams):
704 def _initparams(self, mandatoryparams, advisoryparams):
704 """internal function to setup all logic related parameters"""
705 """internal function to setup all logic related parameters"""
705 # make it read only to prevent people touching it by mistake.
706 # make it read only to prevent people touching it by mistake.
706 self.mandatoryparams = tuple(mandatoryparams)
707 self.mandatoryparams = tuple(mandatoryparams)
707 self.advisoryparams = tuple(advisoryparams)
708 self.advisoryparams = tuple(advisoryparams)
708 # user friendly UI
709 # user friendly UI
709 self.params = dict(self.mandatoryparams)
710 self.params = dict(self.mandatoryparams)
710 self.params.update(dict(self.advisoryparams))
711 self.params.update(dict(self.advisoryparams))
711 self.mandatorykeys = frozenset(p[0] for p in mandatoryparams)
712 self.mandatorykeys = frozenset(p[0] for p in mandatoryparams)
712
713
713 def _readheader(self):
714 def _readheader(self):
714 """read the header and setup the object"""
715 """read the header and setup the object"""
715 typesize = self._unpackheader(_fparttypesize)[0]
716 typesize = self._unpackheader(_fparttypesize)[0]
716 self.type = self._fromheader(typesize)
717 self.type = self._fromheader(typesize)
717 self.ui.debug('part type: "%s"\n' % self.type)
718 self.ui.debug('part type: "%s"\n' % self.type)
718 self.id = self._unpackheader(_fpartid)[0]
719 self.id = self._unpackheader(_fpartid)[0]
719 self.ui.debug('part id: "%s"\n' % self.id)
720 self.ui.debug('part id: "%s"\n' % self.id)
720 ## reading parameters
721 ## reading parameters
721 # param count
722 # param count
722 mancount, advcount = self._unpackheader(_fpartparamcount)
723 mancount, advcount = self._unpackheader(_fpartparamcount)
723 self.ui.debug('part parameters: %i\n' % (mancount + advcount))
724 self.ui.debug('part parameters: %i\n' % (mancount + advcount))
724 # param size
725 # param size
725 fparamsizes = _makefpartparamsizes(mancount + advcount)
726 fparamsizes = _makefpartparamsizes(mancount + advcount)
726 paramsizes = self._unpackheader(fparamsizes)
727 paramsizes = self._unpackheader(fparamsizes)
727 # make it a list of couple again
728 # make it a list of couple again
728 paramsizes = zip(paramsizes[::2], paramsizes[1::2])
729 paramsizes = zip(paramsizes[::2], paramsizes[1::2])
729 # split mandatory from advisory
730 # split mandatory from advisory
730 mansizes = paramsizes[:mancount]
731 mansizes = paramsizes[:mancount]
731 advsizes = paramsizes[mancount:]
732 advsizes = paramsizes[mancount:]
732 # retrive param value
733 # retrive param value
733 manparams = []
734 manparams = []
734 for key, value in mansizes:
735 for key, value in mansizes:
735 manparams.append((self._fromheader(key), self._fromheader(value)))
736 manparams.append((self._fromheader(key), self._fromheader(value)))
736 advparams = []
737 advparams = []
737 for key, value in advsizes:
738 for key, value in advsizes:
738 advparams.append((self._fromheader(key), self._fromheader(value)))
739 advparams.append((self._fromheader(key), self._fromheader(value)))
739 self._initparams(manparams, advparams)
740 self._initparams(manparams, advparams)
740 ## part payload
741 ## part payload
741 def payloadchunks():
742 def payloadchunks():
742 payloadsize = self._unpack(_fpayloadsize)[0]
743 payloadsize = self._unpack(_fpayloadsize)[0]
743 self.ui.debug('payload chunk size: %i\n' % payloadsize)
744 self.ui.debug('payload chunk size: %i\n' % payloadsize)
744 while payloadsize:
745 while payloadsize:
745 yield self._readexact(payloadsize)
746 yield self._readexact(payloadsize)
746 payloadsize = self._unpack(_fpayloadsize)[0]
747 payloadsize = self._unpack(_fpayloadsize)[0]
747 self.ui.debug('payload chunk size: %i\n' % payloadsize)
748 self.ui.debug('payload chunk size: %i\n' % payloadsize)
748 self._payloadstream = util.chunkbuffer(payloadchunks())
749 self._payloadstream = util.chunkbuffer(payloadchunks())
749 # we read the data, tell it
750 # we read the data, tell it
750 self._initialized = True
751 self._initialized = True
751
752
752 def read(self, size=None):
753 def read(self, size=None):
753 """read payload data"""
754 """read payload data"""
754 if not self._initialized:
755 if not self._initialized:
755 self._readheader()
756 self._readheader()
756 if size is None:
757 if size is None:
757 data = self._payloadstream.read()
758 data = self._payloadstream.read()
758 else:
759 else:
759 data = self._payloadstream.read(size)
760 data = self._payloadstream.read(size)
760 if size is None or len(data) < size:
761 if size is None or len(data) < size:
761 self.consumed = True
762 self.consumed = True
762 return data
763 return data
763
764
764
765
765 @parthandler('b2x:changegroup')
766 @parthandler('b2x:changegroup')
766 def handlechangegroup(op, inpart):
767 def handlechangegroup(op, inpart):
767 """apply a changegroup part on the repo
768 """apply a changegroup part on the repo
768
769
769 This is a very early implementation that will massive rework before being
770 This is a very early implementation that will massive rework before being
770 inflicted to any end-user.
771 inflicted to any end-user.
771 """
772 """
772 # Make sure we trigger a transaction creation
773 # Make sure we trigger a transaction creation
773 #
774 #
774 # The addchangegroup function will get a transaction object by itself, but
775 # The addchangegroup function will get a transaction object by itself, but
775 # we need to make sure we trigger the creation of a transaction object used
776 # we need to make sure we trigger the creation of a transaction object used
776 # for the whole processing scope.
777 # for the whole processing scope.
777 op.gettransaction()
778 op.gettransaction()
778 cg = changegroup.unbundle10(inpart, 'UN')
779 cg = changegroup.unbundle10(inpart, 'UN')
779 ret = changegroup.addchangegroup(op.repo, cg, 'bundle2', 'bundle2')
780 ret = changegroup.addchangegroup(op.repo, cg, 'bundle2', 'bundle2')
780 op.records.add('changegroup', {'return': ret})
781 op.records.add('changegroup', {'return': ret})
781 if op.reply is not None:
782 if op.reply is not None:
782 # This is definitly not the final form of this
783 # This is definitly not the final form of this
783 # return. But one need to start somewhere.
784 # return. But one need to start somewhere.
784 part = op.reply.newpart('b2x:reply:changegroup')
785 part = op.reply.newpart('b2x:reply:changegroup')
785 part.addparam('in-reply-to', str(inpart.id), mandatory=False)
786 part.addparam('in-reply-to', str(inpart.id), mandatory=False)
786 part.addparam('return', '%i' % ret, mandatory=False)
787 part.addparam('return', '%i' % ret, mandatory=False)
787 assert not inpart.read()
788 assert not inpart.read()
788
789
789 @parthandler('b2x:reply:changegroup')
790 @parthandler('b2x:reply:changegroup')
790 def handlechangegroup(op, inpart):
791 def handlechangegroup(op, inpart):
791 ret = int(inpart.params['return'])
792 ret = int(inpart.params['return'])
792 replyto = int(inpart.params['in-reply-to'])
793 replyto = int(inpart.params['in-reply-to'])
793 op.records.add('changegroup', {'return': ret}, replyto)
794 op.records.add('changegroup', {'return': ret}, replyto)
794
795
795 @parthandler('b2x:check:heads')
796 @parthandler('b2x:check:heads')
796 def handlechangegroup(op, inpart):
797 def handlechangegroup(op, inpart):
797 """check that head of the repo did not change
798 """check that head of the repo did not change
798
799
799 This is used to detect a push race when using unbundle.
800 This is used to detect a push race when using unbundle.
800 This replaces the "heads" argument of unbundle."""
801 This replaces the "heads" argument of unbundle."""
801 h = inpart.read(20)
802 h = inpart.read(20)
802 heads = []
803 heads = []
803 while len(h) == 20:
804 while len(h) == 20:
804 heads.append(h)
805 heads.append(h)
805 h = inpart.read(20)
806 h = inpart.read(20)
806 assert not h
807 assert not h
807 if heads != op.repo.heads():
808 if heads != op.repo.heads():
808 raise error.PushRaced('repository changed while pushing - '
809 raise error.PushRaced('repository changed while pushing - '
809 'please try again')
810 'please try again')
810
811
811 @parthandler('b2x:output')
812 @parthandler('b2x:output')
812 def handleoutput(op, inpart):
813 def handleoutput(op, inpart):
813 """forward output captured on the server to the client"""
814 """forward output captured on the server to the client"""
814 for line in inpart.read().splitlines():
815 for line in inpart.read().splitlines():
815 op.ui.write(('remote: %s\n' % line))
816 op.ui.write(('remote: %s\n' % line))
816
817
817 @parthandler('b2x:replycaps')
818 @parthandler('b2x:replycaps')
818 def handlereplycaps(op, inpart):
819 def handlereplycaps(op, inpart):
819 """Notify that a reply bundle should be created
820 """Notify that a reply bundle should be created
820
821
821 The payload contains the capabilities information for the reply"""
822 The payload contains the capabilities information for the reply"""
822 caps = decodecaps(inpart.read())
823 caps = decodecaps(inpart.read())
823 if op.reply is None:
824 if op.reply is None:
824 op.reply = bundle20(op.ui, caps)
825 op.reply = bundle20(op.ui, caps)
825
826
826 @parthandler('b2x:error:abort')
827 @parthandler('b2x:error:abort')
827 def handlereplycaps(op, inpart):
828 def handlereplycaps(op, inpart):
828 """Used to transmit abort error over the wire"""
829 """Used to transmit abort error over the wire"""
829 raise util.Abort(inpart.params['message'], hint=inpart.params.get('hint'))
830 raise util.Abort(inpart.params['message'], hint=inpart.params.get('hint'))
830
831
831 @parthandler('b2x:error:unsupportedcontent')
832 @parthandler('b2x:error:unsupportedcontent')
832 def handlereplycaps(op, inpart):
833 def handlereplycaps(op, inpart):
833 """Used to transmit unknown content error over the wire"""
834 """Used to transmit unknown content error over the wire"""
834 kwargs = {}
835 kwargs = {}
835 kwargs['parttype'] = inpart.params['parttype']
836 kwargs['parttype'] = inpart.params['parttype']
836 params = inpart.params.get('params')
837 params = inpart.params.get('params')
837 if params is not None:
838 if params is not None:
838 kwargs['params'] = params.split('\0')
839 kwargs['params'] = params.split('\0')
839
840
840 raise error.BundleValueError(**kwargs)
841 raise error.BundleValueError(**kwargs)
841
842
842 @parthandler('b2x:error:pushraced')
843 @parthandler('b2x:error:pushraced', ('message',))
843 def handlereplycaps(op, inpart):
844 def handlereplycaps(op, inpart):
844 """Used to transmit push race error over the wire"""
845 """Used to transmit push race error over the wire"""
845 raise error.ResponseError(_('push failed:'), inpart.params['message'])
846 raise error.ResponseError(_('push failed:'), inpart.params['message'])
General Comments 0
You need to be logged in to leave comments. Login now