##// END OF EJS Templates
bundle2: add an informative comment to the capability dict...
Pierre-Yves David -
r25317:5a5b7046 default
parent child Browse files
Show More
@@ -1,1276 +1,1278 b''
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: int32
34 :params size: int32
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: int32
67 :header size: int32
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 (restricted to [a-zA-Z0-9_:-]*)
88 :parttype: alphanumerical part name (restricted to [a-zA-Z0-9_:-]*)
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 an int32, `chunkdata` are plain bytes (as much as
122 `chunksize` is an int32, `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 `chunksize` can be negative to trigger special case processing. No such
128 `chunksize` can be negative to trigger special case processing. No such
129 processing is in place yet.
129 processing is in place yet.
130
130
131 Bundle processing
131 Bundle processing
132 ============================
132 ============================
133
133
134 Each part is processed in order using a "part handler". Handler are registered
134 Each part is processed in order using a "part handler". Handler are registered
135 for a certain part type.
135 for a certain part type.
136
136
137 The matching of a part to its handler is case insensitive. The case of the
137 The matching of a part to its handler is case insensitive. The case of the
138 part type is used to know if a part is mandatory or advisory. If the Part type
138 part type is used to know if a part is mandatory or advisory. If the Part type
139 contains any uppercase char it is considered mandatory. When no handler is
139 contains any uppercase char it is considered mandatory. When no handler is
140 known for a Mandatory part, the process is aborted and an exception is raised.
140 known for a Mandatory part, the process is aborted and an exception is raised.
141 If the part is advisory and no handler is known, the part is ignored. When the
141 If the part is advisory and no handler is known, the part is ignored. When the
142 process is aborted, the full bundle is still read from the stream to keep the
142 process is aborted, the full bundle is still read from the stream to keep the
143 channel usable. But none of the part read from an abort are processed. In the
143 channel usable. But none of the part read from an abort are processed. In the
144 future, dropping the stream may become an option for channel we do not care to
144 future, dropping the stream may become an option for channel we do not care to
145 preserve.
145 preserve.
146 """
146 """
147
147
148 import errno
148 import errno
149 import sys
149 import sys
150 import util
150 import util
151 import struct
151 import struct
152 import urllib
152 import urllib
153 import string
153 import string
154 import obsolete
154 import obsolete
155 import pushkey
155 import pushkey
156 import url
156 import url
157 import re
157 import re
158
158
159 import changegroup, error
159 import changegroup, error
160 from i18n import _
160 from i18n import _
161
161
162 _pack = struct.pack
162 _pack = struct.pack
163 _unpack = struct.unpack
163 _unpack = struct.unpack
164
164
165 _fstreamparamsize = '>i'
165 _fstreamparamsize = '>i'
166 _fpartheadersize = '>i'
166 _fpartheadersize = '>i'
167 _fparttypesize = '>B'
167 _fparttypesize = '>B'
168 _fpartid = '>I'
168 _fpartid = '>I'
169 _fpayloadsize = '>i'
169 _fpayloadsize = '>i'
170 _fpartparamcount = '>BB'
170 _fpartparamcount = '>BB'
171
171
172 preferedchunksize = 4096
172 preferedchunksize = 4096
173
173
174 _parttypeforbidden = re.compile('[^a-zA-Z0-9_:-]')
174 _parttypeforbidden = re.compile('[^a-zA-Z0-9_:-]')
175
175
176 def outdebug(ui, message):
176 def outdebug(ui, message):
177 """debug regarding output stream (bundling)"""
177 """debug regarding output stream (bundling)"""
178 ui.debug('bundle2-output: %s\n' % message)
178 ui.debug('bundle2-output: %s\n' % message)
179
179
180 def validateparttype(parttype):
180 def validateparttype(parttype):
181 """raise ValueError if a parttype contains invalid character"""
181 """raise ValueError if a parttype contains invalid character"""
182 if _parttypeforbidden.search(parttype):
182 if _parttypeforbidden.search(parttype):
183 raise ValueError(parttype)
183 raise ValueError(parttype)
184
184
185 def _makefpartparamsizes(nbparams):
185 def _makefpartparamsizes(nbparams):
186 """return a struct format to read part parameter sizes
186 """return a struct format to read part parameter sizes
187
187
188 The number parameters is variable so we need to build that format
188 The number parameters is variable so we need to build that format
189 dynamically.
189 dynamically.
190 """
190 """
191 return '>'+('BB'*nbparams)
191 return '>'+('BB'*nbparams)
192
192
193 parthandlermapping = {}
193 parthandlermapping = {}
194
194
195 def parthandler(parttype, params=()):
195 def parthandler(parttype, params=()):
196 """decorator that register a function as a bundle2 part handler
196 """decorator that register a function as a bundle2 part handler
197
197
198 eg::
198 eg::
199
199
200 @parthandler('myparttype', ('mandatory', 'param', 'handled'))
200 @parthandler('myparttype', ('mandatory', 'param', 'handled'))
201 def myparttypehandler(...):
201 def myparttypehandler(...):
202 '''process a part of type "my part".'''
202 '''process a part of type "my part".'''
203 ...
203 ...
204 """
204 """
205 validateparttype(parttype)
205 validateparttype(parttype)
206 def _decorator(func):
206 def _decorator(func):
207 lparttype = parttype.lower() # enforce lower case matching.
207 lparttype = parttype.lower() # enforce lower case matching.
208 assert lparttype not in parthandlermapping
208 assert lparttype not in parthandlermapping
209 parthandlermapping[lparttype] = func
209 parthandlermapping[lparttype] = func
210 func.params = frozenset(params)
210 func.params = frozenset(params)
211 return func
211 return func
212 return _decorator
212 return _decorator
213
213
214 class unbundlerecords(object):
214 class unbundlerecords(object):
215 """keep record of what happens during and unbundle
215 """keep record of what happens during and unbundle
216
216
217 New records are added using `records.add('cat', obj)`. Where 'cat' is a
217 New records are added using `records.add('cat', obj)`. Where 'cat' is a
218 category of record and obj is an arbitrary object.
218 category of record and obj is an arbitrary object.
219
219
220 `records['cat']` will return all entries of this category 'cat'.
220 `records['cat']` will return all entries of this category 'cat'.
221
221
222 Iterating on the object itself will yield `('category', obj)` tuples
222 Iterating on the object itself will yield `('category', obj)` tuples
223 for all entries.
223 for all entries.
224
224
225 All iterations happens in chronological order.
225 All iterations happens in chronological order.
226 """
226 """
227
227
228 def __init__(self):
228 def __init__(self):
229 self._categories = {}
229 self._categories = {}
230 self._sequences = []
230 self._sequences = []
231 self._replies = {}
231 self._replies = {}
232
232
233 def add(self, category, entry, inreplyto=None):
233 def add(self, category, entry, inreplyto=None):
234 """add a new record of a given category.
234 """add a new record of a given category.
235
235
236 The entry can then be retrieved in the list returned by
236 The entry can then be retrieved in the list returned by
237 self['category']."""
237 self['category']."""
238 self._categories.setdefault(category, []).append(entry)
238 self._categories.setdefault(category, []).append(entry)
239 self._sequences.append((category, entry))
239 self._sequences.append((category, entry))
240 if inreplyto is not None:
240 if inreplyto is not None:
241 self.getreplies(inreplyto).add(category, entry)
241 self.getreplies(inreplyto).add(category, entry)
242
242
243 def getreplies(self, partid):
243 def getreplies(self, partid):
244 """get the records that are replies to a specific part"""
244 """get the records that are replies to a specific part"""
245 return self._replies.setdefault(partid, unbundlerecords())
245 return self._replies.setdefault(partid, unbundlerecords())
246
246
247 def __getitem__(self, cat):
247 def __getitem__(self, cat):
248 return tuple(self._categories.get(cat, ()))
248 return tuple(self._categories.get(cat, ()))
249
249
250 def __iter__(self):
250 def __iter__(self):
251 return iter(self._sequences)
251 return iter(self._sequences)
252
252
253 def __len__(self):
253 def __len__(self):
254 return len(self._sequences)
254 return len(self._sequences)
255
255
256 def __nonzero__(self):
256 def __nonzero__(self):
257 return bool(self._sequences)
257 return bool(self._sequences)
258
258
259 class bundleoperation(object):
259 class bundleoperation(object):
260 """an object that represents a single bundling process
260 """an object that represents a single bundling process
261
261
262 Its purpose is to carry unbundle-related objects and states.
262 Its purpose is to carry unbundle-related objects and states.
263
263
264 A new object should be created at the beginning of each bundle processing.
264 A new object should be created at the beginning of each bundle processing.
265 The object is to be returned by the processing function.
265 The object is to be returned by the processing function.
266
266
267 The object has very little content now it will ultimately contain:
267 The object has very little content now it will ultimately contain:
268 * an access to the repo the bundle is applied to,
268 * an access to the repo the bundle is applied to,
269 * a ui object,
269 * a ui object,
270 * a way to retrieve a transaction to add changes to the repo,
270 * a way to retrieve a transaction to add changes to the repo,
271 * a way to record the result of processing each part,
271 * a way to record the result of processing each part,
272 * a way to construct a bundle response when applicable.
272 * a way to construct a bundle response when applicable.
273 """
273 """
274
274
275 def __init__(self, repo, transactiongetter, captureoutput=True):
275 def __init__(self, repo, transactiongetter, captureoutput=True):
276 self.repo = repo
276 self.repo = repo
277 self.ui = repo.ui
277 self.ui = repo.ui
278 self.records = unbundlerecords()
278 self.records = unbundlerecords()
279 self.gettransaction = transactiongetter
279 self.gettransaction = transactiongetter
280 self.reply = None
280 self.reply = None
281 self.captureoutput = captureoutput
281 self.captureoutput = captureoutput
282
282
283 class TransactionUnavailable(RuntimeError):
283 class TransactionUnavailable(RuntimeError):
284 pass
284 pass
285
285
286 def _notransaction():
286 def _notransaction():
287 """default method to get a transaction while processing a bundle
287 """default method to get a transaction while processing a bundle
288
288
289 Raise an exception to highlight the fact that no transaction was expected
289 Raise an exception to highlight the fact that no transaction was expected
290 to be created"""
290 to be created"""
291 raise TransactionUnavailable()
291 raise TransactionUnavailable()
292
292
293 def processbundle(repo, unbundler, transactiongetter=None, op=None):
293 def processbundle(repo, unbundler, transactiongetter=None, op=None):
294 """This function process a bundle, apply effect to/from a repo
294 """This function process a bundle, apply effect to/from a repo
295
295
296 It iterates over each part then searches for and uses the proper handling
296 It iterates over each part then searches for and uses the proper handling
297 code to process the part. Parts are processed in order.
297 code to process the part. Parts are processed in order.
298
298
299 This is very early version of this function that will be strongly reworked
299 This is very early version of this function that will be strongly reworked
300 before final usage.
300 before final usage.
301
301
302 Unknown Mandatory part will abort the process.
302 Unknown Mandatory part will abort the process.
303
303
304 It is temporarily possible to provide a prebuilt bundleoperation to the
304 It is temporarily possible to provide a prebuilt bundleoperation to the
305 function. This is used to ensure output is properly propagated in case of
305 function. This is used to ensure output is properly propagated in case of
306 an error during the unbundling. This output capturing part will likely be
306 an error during the unbundling. This output capturing part will likely be
307 reworked and this ability will probably go away in the process.
307 reworked and this ability will probably go away in the process.
308 """
308 """
309 if op is None:
309 if op is None:
310 if transactiongetter is None:
310 if transactiongetter is None:
311 transactiongetter = _notransaction
311 transactiongetter = _notransaction
312 op = bundleoperation(repo, transactiongetter)
312 op = bundleoperation(repo, transactiongetter)
313 # todo:
313 # todo:
314 # - replace this is a init function soon.
314 # - replace this is a init function soon.
315 # - exception catching
315 # - exception catching
316 unbundler.params
316 unbundler.params
317 iterparts = unbundler.iterparts()
317 iterparts = unbundler.iterparts()
318 part = None
318 part = None
319 try:
319 try:
320 for part in iterparts:
320 for part in iterparts:
321 _processpart(op, part)
321 _processpart(op, part)
322 except BaseException, exc:
322 except BaseException, exc:
323 for part in iterparts:
323 for part in iterparts:
324 # consume the bundle content
324 # consume the bundle content
325 part.seek(0, 2)
325 part.seek(0, 2)
326 # Small hack to let caller code distinguish exceptions from bundle2
326 # Small hack to let caller code distinguish exceptions from bundle2
327 # processing from processing the old format. This is mostly
327 # processing from processing the old format. This is mostly
328 # needed to handle different return codes to unbundle according to the
328 # needed to handle different return codes to unbundle according to the
329 # type of bundle. We should probably clean up or drop this return code
329 # type of bundle. We should probably clean up or drop this return code
330 # craziness in a future version.
330 # craziness in a future version.
331 exc.duringunbundle2 = True
331 exc.duringunbundle2 = True
332 salvaged = []
332 salvaged = []
333 if op.reply is not None:
333 if op.reply is not None:
334 salvaged = op.reply.salvageoutput()
334 salvaged = op.reply.salvageoutput()
335 exc._bundle2salvagedoutput = salvaged
335 exc._bundle2salvagedoutput = salvaged
336 raise
336 raise
337 return op
337 return op
338
338
339 def _processpart(op, part):
339 def _processpart(op, part):
340 """process a single part from a bundle
340 """process a single part from a bundle
341
341
342 The part is guaranteed to have been fully consumed when the function exits
342 The part is guaranteed to have been fully consumed when the function exits
343 (even if an exception is raised)."""
343 (even if an exception is raised)."""
344 try:
344 try:
345 try:
345 try:
346 handler = parthandlermapping.get(part.type)
346 handler = parthandlermapping.get(part.type)
347 if handler is None:
347 if handler is None:
348 raise error.UnsupportedPartError(parttype=part.type)
348 raise error.UnsupportedPartError(parttype=part.type)
349 op.ui.debug('found a handler for part %r\n' % part.type)
349 op.ui.debug('found a handler for part %r\n' % part.type)
350 unknownparams = part.mandatorykeys - handler.params
350 unknownparams = part.mandatorykeys - handler.params
351 if unknownparams:
351 if unknownparams:
352 unknownparams = list(unknownparams)
352 unknownparams = list(unknownparams)
353 unknownparams.sort()
353 unknownparams.sort()
354 raise error.UnsupportedPartError(parttype=part.type,
354 raise error.UnsupportedPartError(parttype=part.type,
355 params=unknownparams)
355 params=unknownparams)
356 except error.UnsupportedPartError, exc:
356 except error.UnsupportedPartError, exc:
357 if part.mandatory: # mandatory parts
357 if part.mandatory: # mandatory parts
358 raise
358 raise
359 op.ui.debug('ignoring unsupported advisory part %s\n' % exc)
359 op.ui.debug('ignoring unsupported advisory part %s\n' % exc)
360 return # skip to part processing
360 return # skip to part processing
361
361
362 # handler is called outside the above try block so that we don't
362 # handler is called outside the above try block so that we don't
363 # risk catching KeyErrors from anything other than the
363 # risk catching KeyErrors from anything other than the
364 # parthandlermapping lookup (any KeyError raised by handler()
364 # parthandlermapping lookup (any KeyError raised by handler()
365 # itself represents a defect of a different variety).
365 # itself represents a defect of a different variety).
366 output = None
366 output = None
367 if op.captureoutput and op.reply is not None:
367 if op.captureoutput and op.reply is not None:
368 op.ui.pushbuffer(error=True, subproc=True)
368 op.ui.pushbuffer(error=True, subproc=True)
369 output = ''
369 output = ''
370 try:
370 try:
371 handler(op, part)
371 handler(op, part)
372 finally:
372 finally:
373 if output is not None:
373 if output is not None:
374 output = op.ui.popbuffer()
374 output = op.ui.popbuffer()
375 if output:
375 if output:
376 outpart = op.reply.newpart('output', data=output,
376 outpart = op.reply.newpart('output', data=output,
377 mandatory=False)
377 mandatory=False)
378 outpart.addparam('in-reply-to', str(part.id), mandatory=False)
378 outpart.addparam('in-reply-to', str(part.id), mandatory=False)
379 finally:
379 finally:
380 # consume the part content to not corrupt the stream.
380 # consume the part content to not corrupt the stream.
381 part.seek(0, 2)
381 part.seek(0, 2)
382
382
383
383
384 def decodecaps(blob):
384 def decodecaps(blob):
385 """decode a bundle2 caps bytes blob into a dictionary
385 """decode a bundle2 caps bytes blob into a dictionary
386
386
387 The blob is a list of capabilities (one per line)
387 The blob is a list of capabilities (one per line)
388 Capabilities may have values using a line of the form::
388 Capabilities may have values using a line of the form::
389
389
390 capability=value1,value2,value3
390 capability=value1,value2,value3
391
391
392 The values are always a list."""
392 The values are always a list."""
393 caps = {}
393 caps = {}
394 for line in blob.splitlines():
394 for line in blob.splitlines():
395 if not line:
395 if not line:
396 continue
396 continue
397 if '=' not in line:
397 if '=' not in line:
398 key, vals = line, ()
398 key, vals = line, ()
399 else:
399 else:
400 key, vals = line.split('=', 1)
400 key, vals = line.split('=', 1)
401 vals = vals.split(',')
401 vals = vals.split(',')
402 key = urllib.unquote(key)
402 key = urllib.unquote(key)
403 vals = [urllib.unquote(v) for v in vals]
403 vals = [urllib.unquote(v) for v in vals]
404 caps[key] = vals
404 caps[key] = vals
405 return caps
405 return caps
406
406
407 def encodecaps(caps):
407 def encodecaps(caps):
408 """encode a bundle2 caps dictionary into a bytes blob"""
408 """encode a bundle2 caps dictionary into a bytes blob"""
409 chunks = []
409 chunks = []
410 for ca in sorted(caps):
410 for ca in sorted(caps):
411 vals = caps[ca]
411 vals = caps[ca]
412 ca = urllib.quote(ca)
412 ca = urllib.quote(ca)
413 vals = [urllib.quote(v) for v in vals]
413 vals = [urllib.quote(v) for v in vals]
414 if vals:
414 if vals:
415 ca = "%s=%s" % (ca, ','.join(vals))
415 ca = "%s=%s" % (ca, ','.join(vals))
416 chunks.append(ca)
416 chunks.append(ca)
417 return '\n'.join(chunks)
417 return '\n'.join(chunks)
418
418
419 class bundle20(object):
419 class bundle20(object):
420 """represent an outgoing bundle2 container
420 """represent an outgoing bundle2 container
421
421
422 Use the `addparam` method to add stream level parameter. and `newpart` to
422 Use the `addparam` method to add stream level parameter. and `newpart` to
423 populate it. Then call `getchunks` to retrieve all the binary chunks of
423 populate it. Then call `getchunks` to retrieve all the binary chunks of
424 data that compose the bundle2 container."""
424 data that compose the bundle2 container."""
425
425
426 _magicstring = 'HG20'
426 _magicstring = 'HG20'
427
427
428 def __init__(self, ui, capabilities=()):
428 def __init__(self, ui, capabilities=()):
429 self.ui = ui
429 self.ui = ui
430 self._params = []
430 self._params = []
431 self._parts = []
431 self._parts = []
432 self.capabilities = dict(capabilities)
432 self.capabilities = dict(capabilities)
433
433
434 @property
434 @property
435 def nbparts(self):
435 def nbparts(self):
436 """total number of parts added to the bundler"""
436 """total number of parts added to the bundler"""
437 return len(self._parts)
437 return len(self._parts)
438
438
439 # methods used to defines the bundle2 content
439 # methods used to defines the bundle2 content
440 def addparam(self, name, value=None):
440 def addparam(self, name, value=None):
441 """add a stream level parameter"""
441 """add a stream level parameter"""
442 if not name:
442 if not name:
443 raise ValueError('empty parameter name')
443 raise ValueError('empty parameter name')
444 if name[0] not in string.letters:
444 if name[0] not in string.letters:
445 raise ValueError('non letter first character: %r' % name)
445 raise ValueError('non letter first character: %r' % name)
446 self._params.append((name, value))
446 self._params.append((name, value))
447
447
448 def addpart(self, part):
448 def addpart(self, part):
449 """add a new part to the bundle2 container
449 """add a new part to the bundle2 container
450
450
451 Parts contains the actual applicative payload."""
451 Parts contains the actual applicative payload."""
452 assert part.id is None
452 assert part.id is None
453 part.id = len(self._parts) # very cheap counter
453 part.id = len(self._parts) # very cheap counter
454 self._parts.append(part)
454 self._parts.append(part)
455
455
456 def newpart(self, typeid, *args, **kwargs):
456 def newpart(self, typeid, *args, **kwargs):
457 """create a new part and add it to the containers
457 """create a new part and add it to the containers
458
458
459 As the part is directly added to the containers. For now, this means
459 As the part is directly added to the containers. For now, this means
460 that any failure to properly initialize the part after calling
460 that any failure to properly initialize the part after calling
461 ``newpart`` should result in a failure of the whole bundling process.
461 ``newpart`` should result in a failure of the whole bundling process.
462
462
463 You can still fall back to manually create and add if you need better
463 You can still fall back to manually create and add if you need better
464 control."""
464 control."""
465 part = bundlepart(typeid, *args, **kwargs)
465 part = bundlepart(typeid, *args, **kwargs)
466 self.addpart(part)
466 self.addpart(part)
467 return part
467 return part
468
468
469 # methods used to generate the bundle2 stream
469 # methods used to generate the bundle2 stream
470 def getchunks(self):
470 def getchunks(self):
471 outdebug(self.ui, 'start emission of %s stream' % self._magicstring)
471 outdebug(self.ui, 'start emission of %s stream' % self._magicstring)
472 yield self._magicstring
472 yield self._magicstring
473 param = self._paramchunk()
473 param = self._paramchunk()
474 outdebug(self.ui, 'bundle parameter: %s' % param)
474 outdebug(self.ui, 'bundle parameter: %s' % param)
475 yield _pack(_fstreamparamsize, len(param))
475 yield _pack(_fstreamparamsize, len(param))
476 if param:
476 if param:
477 yield param
477 yield param
478
478
479 outdebug(self.ui, 'start of parts')
479 outdebug(self.ui, 'start of parts')
480 for part in self._parts:
480 for part in self._parts:
481 outdebug(self.ui, 'bundle part: "%s"' % part.type)
481 outdebug(self.ui, 'bundle part: "%s"' % part.type)
482 for chunk in part.getchunks():
482 for chunk in part.getchunks():
483 yield chunk
483 yield chunk
484 outdebug(self.ui, 'end of bundle')
484 outdebug(self.ui, 'end of bundle')
485 yield _pack(_fpartheadersize, 0)
485 yield _pack(_fpartheadersize, 0)
486
486
487 def _paramchunk(self):
487 def _paramchunk(self):
488 """return a encoded version of all stream parameters"""
488 """return a encoded version of all stream parameters"""
489 blocks = []
489 blocks = []
490 for par, value in self._params:
490 for par, value in self._params:
491 par = urllib.quote(par)
491 par = urllib.quote(par)
492 if value is not None:
492 if value is not None:
493 value = urllib.quote(value)
493 value = urllib.quote(value)
494 par = '%s=%s' % (par, value)
494 par = '%s=%s' % (par, value)
495 blocks.append(par)
495 blocks.append(par)
496 return ' '.join(blocks)
496 return ' '.join(blocks)
497
497
498 def salvageoutput(self):
498 def salvageoutput(self):
499 """return a list with a copy of all output parts in the bundle
499 """return a list with a copy of all output parts in the bundle
500
500
501 This is meant to be used during error handling to make sure we preserve
501 This is meant to be used during error handling to make sure we preserve
502 server output"""
502 server output"""
503 salvaged = []
503 salvaged = []
504 for part in self._parts:
504 for part in self._parts:
505 if part.type.startswith('output'):
505 if part.type.startswith('output'):
506 salvaged.append(part.copy())
506 salvaged.append(part.copy())
507 return salvaged
507 return salvaged
508
508
509
509
510 class unpackermixin(object):
510 class unpackermixin(object):
511 """A mixin to extract bytes and struct data from a stream"""
511 """A mixin to extract bytes and struct data from a stream"""
512
512
513 def __init__(self, fp):
513 def __init__(self, fp):
514 self._fp = fp
514 self._fp = fp
515 self._seekable = (util.safehasattr(fp, 'seek') and
515 self._seekable = (util.safehasattr(fp, 'seek') and
516 util.safehasattr(fp, 'tell'))
516 util.safehasattr(fp, 'tell'))
517
517
518 def _unpack(self, format):
518 def _unpack(self, format):
519 """unpack this struct format from the stream"""
519 """unpack this struct format from the stream"""
520 data = self._readexact(struct.calcsize(format))
520 data = self._readexact(struct.calcsize(format))
521 return _unpack(format, data)
521 return _unpack(format, data)
522
522
523 def _readexact(self, size):
523 def _readexact(self, size):
524 """read exactly <size> bytes from the stream"""
524 """read exactly <size> bytes from the stream"""
525 return changegroup.readexactly(self._fp, size)
525 return changegroup.readexactly(self._fp, size)
526
526
527 def seek(self, offset, whence=0):
527 def seek(self, offset, whence=0):
528 """move the underlying file pointer"""
528 """move the underlying file pointer"""
529 if self._seekable:
529 if self._seekable:
530 return self._fp.seek(offset, whence)
530 return self._fp.seek(offset, whence)
531 else:
531 else:
532 raise NotImplementedError(_('File pointer is not seekable'))
532 raise NotImplementedError(_('File pointer is not seekable'))
533
533
534 def tell(self):
534 def tell(self):
535 """return the file offset, or None if file is not seekable"""
535 """return the file offset, or None if file is not seekable"""
536 if self._seekable:
536 if self._seekable:
537 try:
537 try:
538 return self._fp.tell()
538 return self._fp.tell()
539 except IOError, e:
539 except IOError, e:
540 if e.errno == errno.ESPIPE:
540 if e.errno == errno.ESPIPE:
541 self._seekable = False
541 self._seekable = False
542 else:
542 else:
543 raise
543 raise
544 return None
544 return None
545
545
546 def close(self):
546 def close(self):
547 """close underlying file"""
547 """close underlying file"""
548 if util.safehasattr(self._fp, 'close'):
548 if util.safehasattr(self._fp, 'close'):
549 return self._fp.close()
549 return self._fp.close()
550
550
551 def getunbundler(ui, fp, header=None):
551 def getunbundler(ui, fp, header=None):
552 """return a valid unbundler object for a given header"""
552 """return a valid unbundler object for a given header"""
553 if header is None:
553 if header is None:
554 header = changegroup.readexactly(fp, 4)
554 header = changegroup.readexactly(fp, 4)
555 magic, version = header[0:2], header[2:4]
555 magic, version = header[0:2], header[2:4]
556 if magic != 'HG':
556 if magic != 'HG':
557 raise util.Abort(_('not a Mercurial bundle'))
557 raise util.Abort(_('not a Mercurial bundle'))
558 unbundlerclass = formatmap.get(version)
558 unbundlerclass = formatmap.get(version)
559 if unbundlerclass is None:
559 if unbundlerclass is None:
560 raise util.Abort(_('unknown bundle version %s') % version)
560 raise util.Abort(_('unknown bundle version %s') % version)
561 unbundler = unbundlerclass(ui, fp)
561 unbundler = unbundlerclass(ui, fp)
562 ui.debug('start processing of %s stream\n' % header)
562 ui.debug('start processing of %s stream\n' % header)
563 return unbundler
563 return unbundler
564
564
565 class unbundle20(unpackermixin):
565 class unbundle20(unpackermixin):
566 """interpret a bundle2 stream
566 """interpret a bundle2 stream
567
567
568 This class is fed with a binary stream and yields parts through its
568 This class is fed with a binary stream and yields parts through its
569 `iterparts` methods."""
569 `iterparts` methods."""
570
570
571 def __init__(self, ui, fp):
571 def __init__(self, ui, fp):
572 """If header is specified, we do not read it out of the stream."""
572 """If header is specified, we do not read it out of the stream."""
573 self.ui = ui
573 self.ui = ui
574 super(unbundle20, self).__init__(fp)
574 super(unbundle20, self).__init__(fp)
575
575
576 @util.propertycache
576 @util.propertycache
577 def params(self):
577 def params(self):
578 """dictionary of stream level parameters"""
578 """dictionary of stream level parameters"""
579 self.ui.debug('reading bundle2 stream parameters\n')
579 self.ui.debug('reading bundle2 stream parameters\n')
580 params = {}
580 params = {}
581 paramssize = self._unpack(_fstreamparamsize)[0]
581 paramssize = self._unpack(_fstreamparamsize)[0]
582 if paramssize < 0:
582 if paramssize < 0:
583 raise error.BundleValueError('negative bundle param size: %i'
583 raise error.BundleValueError('negative bundle param size: %i'
584 % paramssize)
584 % paramssize)
585 if paramssize:
585 if paramssize:
586 for p in self._readexact(paramssize).split(' '):
586 for p in self._readexact(paramssize).split(' '):
587 p = p.split('=', 1)
587 p = p.split('=', 1)
588 p = [urllib.unquote(i) for i in p]
588 p = [urllib.unquote(i) for i in p]
589 if len(p) < 2:
589 if len(p) < 2:
590 p.append(None)
590 p.append(None)
591 self._processparam(*p)
591 self._processparam(*p)
592 params[p[0]] = p[1]
592 params[p[0]] = p[1]
593 return params
593 return params
594
594
595 def _processparam(self, name, value):
595 def _processparam(self, name, value):
596 """process a parameter, applying its effect if needed
596 """process a parameter, applying its effect if needed
597
597
598 Parameter starting with a lower case letter are advisory and will be
598 Parameter starting with a lower case letter are advisory and will be
599 ignored when unknown. Those starting with an upper case letter are
599 ignored when unknown. Those starting with an upper case letter are
600 mandatory and will this function will raise a KeyError when unknown.
600 mandatory and will this function will raise a KeyError when unknown.
601
601
602 Note: no option are currently supported. Any input will be either
602 Note: no option are currently supported. Any input will be either
603 ignored or failing.
603 ignored or failing.
604 """
604 """
605 if not name:
605 if not name:
606 raise ValueError('empty parameter name')
606 raise ValueError('empty parameter name')
607 if name[0] not in string.letters:
607 if name[0] not in string.letters:
608 raise ValueError('non letter first character: %r' % name)
608 raise ValueError('non letter first character: %r' % name)
609 # Some logic will be later added here to try to process the option for
609 # Some logic will be later added here to try to process the option for
610 # a dict of known parameter.
610 # a dict of known parameter.
611 if name[0].islower():
611 if name[0].islower():
612 self.ui.debug("ignoring unknown parameter %r\n" % name)
612 self.ui.debug("ignoring unknown parameter %r\n" % name)
613 else:
613 else:
614 raise error.UnsupportedPartError(params=(name,))
614 raise error.UnsupportedPartError(params=(name,))
615
615
616
616
617 def iterparts(self):
617 def iterparts(self):
618 """yield all parts contained in the stream"""
618 """yield all parts contained in the stream"""
619 # make sure param have been loaded
619 # make sure param have been loaded
620 self.params
620 self.params
621 self.ui.debug('start extraction of bundle2 parts\n')
621 self.ui.debug('start extraction of bundle2 parts\n')
622 headerblock = self._readpartheader()
622 headerblock = self._readpartheader()
623 while headerblock is not None:
623 while headerblock is not None:
624 part = unbundlepart(self.ui, headerblock, self._fp)
624 part = unbundlepart(self.ui, headerblock, self._fp)
625 yield part
625 yield part
626 part.seek(0, 2)
626 part.seek(0, 2)
627 headerblock = self._readpartheader()
627 headerblock = self._readpartheader()
628 self.ui.debug('end of bundle2 stream\n')
628 self.ui.debug('end of bundle2 stream\n')
629
629
630 def _readpartheader(self):
630 def _readpartheader(self):
631 """reads a part header size and return the bytes blob
631 """reads a part header size and return the bytes blob
632
632
633 returns None if empty"""
633 returns None if empty"""
634 headersize = self._unpack(_fpartheadersize)[0]
634 headersize = self._unpack(_fpartheadersize)[0]
635 if headersize < 0:
635 if headersize < 0:
636 raise error.BundleValueError('negative part header size: %i'
636 raise error.BundleValueError('negative part header size: %i'
637 % headersize)
637 % headersize)
638 self.ui.debug('part header size: %i\n' % headersize)
638 self.ui.debug('part header size: %i\n' % headersize)
639 if headersize:
639 if headersize:
640 return self._readexact(headersize)
640 return self._readexact(headersize)
641 return None
641 return None
642
642
643 def compressed(self):
643 def compressed(self):
644 return False
644 return False
645
645
646 formatmap = {'20': unbundle20}
646 formatmap = {'20': unbundle20}
647
647
648 class bundlepart(object):
648 class bundlepart(object):
649 """A bundle2 part contains application level payload
649 """A bundle2 part contains application level payload
650
650
651 The part `type` is used to route the part to the application level
651 The part `type` is used to route the part to the application level
652 handler.
652 handler.
653
653
654 The part payload is contained in ``part.data``. It could be raw bytes or a
654 The part payload is contained in ``part.data``. It could be raw bytes or a
655 generator of byte chunks.
655 generator of byte chunks.
656
656
657 You can add parameters to the part using the ``addparam`` method.
657 You can add parameters to the part using the ``addparam`` method.
658 Parameters can be either mandatory (default) or advisory. Remote side
658 Parameters can be either mandatory (default) or advisory. Remote side
659 should be able to safely ignore the advisory ones.
659 should be able to safely ignore the advisory ones.
660
660
661 Both data and parameters cannot be modified after the generation has begun.
661 Both data and parameters cannot be modified after the generation has begun.
662 """
662 """
663
663
664 def __init__(self, parttype, mandatoryparams=(), advisoryparams=(),
664 def __init__(self, parttype, mandatoryparams=(), advisoryparams=(),
665 data='', mandatory=True):
665 data='', mandatory=True):
666 validateparttype(parttype)
666 validateparttype(parttype)
667 self.id = None
667 self.id = None
668 self.type = parttype
668 self.type = parttype
669 self._data = data
669 self._data = data
670 self._mandatoryparams = list(mandatoryparams)
670 self._mandatoryparams = list(mandatoryparams)
671 self._advisoryparams = list(advisoryparams)
671 self._advisoryparams = list(advisoryparams)
672 # checking for duplicated entries
672 # checking for duplicated entries
673 self._seenparams = set()
673 self._seenparams = set()
674 for pname, __ in self._mandatoryparams + self._advisoryparams:
674 for pname, __ in self._mandatoryparams + self._advisoryparams:
675 if pname in self._seenparams:
675 if pname in self._seenparams:
676 raise RuntimeError('duplicated params: %s' % pname)
676 raise RuntimeError('duplicated params: %s' % pname)
677 self._seenparams.add(pname)
677 self._seenparams.add(pname)
678 # status of the part's generation:
678 # status of the part's generation:
679 # - None: not started,
679 # - None: not started,
680 # - False: currently generated,
680 # - False: currently generated,
681 # - True: generation done.
681 # - True: generation done.
682 self._generated = None
682 self._generated = None
683 self.mandatory = mandatory
683 self.mandatory = mandatory
684
684
685 def copy(self):
685 def copy(self):
686 """return a copy of the part
686 """return a copy of the part
687
687
688 The new part have the very same content but no partid assigned yet.
688 The new part have the very same content but no partid assigned yet.
689 Parts with generated data cannot be copied."""
689 Parts with generated data cannot be copied."""
690 assert not util.safehasattr(self.data, 'next')
690 assert not util.safehasattr(self.data, 'next')
691 return self.__class__(self.type, self._mandatoryparams,
691 return self.__class__(self.type, self._mandatoryparams,
692 self._advisoryparams, self._data, self.mandatory)
692 self._advisoryparams, self._data, self.mandatory)
693
693
694 # methods used to defines the part content
694 # methods used to defines the part content
695 def __setdata(self, data):
695 def __setdata(self, data):
696 if self._generated is not None:
696 if self._generated is not None:
697 raise error.ReadOnlyPartError('part is being generated')
697 raise error.ReadOnlyPartError('part is being generated')
698 self._data = data
698 self._data = data
699 def __getdata(self):
699 def __getdata(self):
700 return self._data
700 return self._data
701 data = property(__getdata, __setdata)
701 data = property(__getdata, __setdata)
702
702
703 @property
703 @property
704 def mandatoryparams(self):
704 def mandatoryparams(self):
705 # make it an immutable tuple to force people through ``addparam``
705 # make it an immutable tuple to force people through ``addparam``
706 return tuple(self._mandatoryparams)
706 return tuple(self._mandatoryparams)
707
707
708 @property
708 @property
709 def advisoryparams(self):
709 def advisoryparams(self):
710 # make it an immutable tuple to force people through ``addparam``
710 # make it an immutable tuple to force people through ``addparam``
711 return tuple(self._advisoryparams)
711 return tuple(self._advisoryparams)
712
712
713 def addparam(self, name, value='', mandatory=True):
713 def addparam(self, name, value='', mandatory=True):
714 if self._generated is not None:
714 if self._generated is not None:
715 raise error.ReadOnlyPartError('part is being generated')
715 raise error.ReadOnlyPartError('part is being generated')
716 if name in self._seenparams:
716 if name in self._seenparams:
717 raise ValueError('duplicated params: %s' % name)
717 raise ValueError('duplicated params: %s' % name)
718 self._seenparams.add(name)
718 self._seenparams.add(name)
719 params = self._advisoryparams
719 params = self._advisoryparams
720 if mandatory:
720 if mandatory:
721 params = self._mandatoryparams
721 params = self._mandatoryparams
722 params.append((name, value))
722 params.append((name, value))
723
723
724 # methods used to generates the bundle2 stream
724 # methods used to generates the bundle2 stream
725 def getchunks(self):
725 def getchunks(self):
726 if self._generated is not None:
726 if self._generated is not None:
727 raise RuntimeError('part can only be consumed once')
727 raise RuntimeError('part can only be consumed once')
728 self._generated = False
728 self._generated = False
729 #### header
729 #### header
730 if self.mandatory:
730 if self.mandatory:
731 parttype = self.type.upper()
731 parttype = self.type.upper()
732 else:
732 else:
733 parttype = self.type.lower()
733 parttype = self.type.lower()
734 ## parttype
734 ## parttype
735 header = [_pack(_fparttypesize, len(parttype)),
735 header = [_pack(_fparttypesize, len(parttype)),
736 parttype, _pack(_fpartid, self.id),
736 parttype, _pack(_fpartid, self.id),
737 ]
737 ]
738 ## parameters
738 ## parameters
739 # count
739 # count
740 manpar = self.mandatoryparams
740 manpar = self.mandatoryparams
741 advpar = self.advisoryparams
741 advpar = self.advisoryparams
742 header.append(_pack(_fpartparamcount, len(manpar), len(advpar)))
742 header.append(_pack(_fpartparamcount, len(manpar), len(advpar)))
743 # size
743 # size
744 parsizes = []
744 parsizes = []
745 for key, value in manpar:
745 for key, value in manpar:
746 parsizes.append(len(key))
746 parsizes.append(len(key))
747 parsizes.append(len(value))
747 parsizes.append(len(value))
748 for key, value in advpar:
748 for key, value in advpar:
749 parsizes.append(len(key))
749 parsizes.append(len(key))
750 parsizes.append(len(value))
750 parsizes.append(len(value))
751 paramsizes = _pack(_makefpartparamsizes(len(parsizes) / 2), *parsizes)
751 paramsizes = _pack(_makefpartparamsizes(len(parsizes) / 2), *parsizes)
752 header.append(paramsizes)
752 header.append(paramsizes)
753 # key, value
753 # key, value
754 for key, value in manpar:
754 for key, value in manpar:
755 header.append(key)
755 header.append(key)
756 header.append(value)
756 header.append(value)
757 for key, value in advpar:
757 for key, value in advpar:
758 header.append(key)
758 header.append(key)
759 header.append(value)
759 header.append(value)
760 ## finalize header
760 ## finalize header
761 headerchunk = ''.join(header)
761 headerchunk = ''.join(header)
762 yield _pack(_fpartheadersize, len(headerchunk))
762 yield _pack(_fpartheadersize, len(headerchunk))
763 yield headerchunk
763 yield headerchunk
764 ## payload
764 ## payload
765 try:
765 try:
766 for chunk in self._payloadchunks():
766 for chunk in self._payloadchunks():
767 yield _pack(_fpayloadsize, len(chunk))
767 yield _pack(_fpayloadsize, len(chunk))
768 yield chunk
768 yield chunk
769 except BaseException, exc:
769 except BaseException, exc:
770 # backup exception data for later
770 # backup exception data for later
771 exc_info = sys.exc_info()
771 exc_info = sys.exc_info()
772 msg = 'unexpected error: %s' % exc
772 msg = 'unexpected error: %s' % exc
773 interpart = bundlepart('error:abort', [('message', msg)],
773 interpart = bundlepart('error:abort', [('message', msg)],
774 mandatory=False)
774 mandatory=False)
775 interpart.id = 0
775 interpart.id = 0
776 yield _pack(_fpayloadsize, -1)
776 yield _pack(_fpayloadsize, -1)
777 for chunk in interpart.getchunks():
777 for chunk in interpart.getchunks():
778 yield chunk
778 yield chunk
779 # abort current part payload
779 # abort current part payload
780 yield _pack(_fpayloadsize, 0)
780 yield _pack(_fpayloadsize, 0)
781 raise exc_info[0], exc_info[1], exc_info[2]
781 raise exc_info[0], exc_info[1], exc_info[2]
782 # end of payload
782 # end of payload
783 yield _pack(_fpayloadsize, 0)
783 yield _pack(_fpayloadsize, 0)
784 self._generated = True
784 self._generated = True
785
785
786 def _payloadchunks(self):
786 def _payloadchunks(self):
787 """yield chunks of a the part payload
787 """yield chunks of a the part payload
788
788
789 Exists to handle the different methods to provide data to a part."""
789 Exists to handle the different methods to provide data to a part."""
790 # we only support fixed size data now.
790 # we only support fixed size data now.
791 # This will be improved in the future.
791 # This will be improved in the future.
792 if util.safehasattr(self.data, 'next'):
792 if util.safehasattr(self.data, 'next'):
793 buff = util.chunkbuffer(self.data)
793 buff = util.chunkbuffer(self.data)
794 chunk = buff.read(preferedchunksize)
794 chunk = buff.read(preferedchunksize)
795 while chunk:
795 while chunk:
796 yield chunk
796 yield chunk
797 chunk = buff.read(preferedchunksize)
797 chunk = buff.read(preferedchunksize)
798 elif len(self.data):
798 elif len(self.data):
799 yield self.data
799 yield self.data
800
800
801
801
802 flaginterrupt = -1
802 flaginterrupt = -1
803
803
804 class interrupthandler(unpackermixin):
804 class interrupthandler(unpackermixin):
805 """read one part and process it with restricted capability
805 """read one part and process it with restricted capability
806
806
807 This allows to transmit exception raised on the producer size during part
807 This allows to transmit exception raised on the producer size during part
808 iteration while the consumer is reading a part.
808 iteration while the consumer is reading a part.
809
809
810 Part processed in this manner only have access to a ui object,"""
810 Part processed in this manner only have access to a ui object,"""
811
811
812 def __init__(self, ui, fp):
812 def __init__(self, ui, fp):
813 super(interrupthandler, self).__init__(fp)
813 super(interrupthandler, self).__init__(fp)
814 self.ui = ui
814 self.ui = ui
815
815
816 def _readpartheader(self):
816 def _readpartheader(self):
817 """reads a part header size and return the bytes blob
817 """reads a part header size and return the bytes blob
818
818
819 returns None if empty"""
819 returns None if empty"""
820 headersize = self._unpack(_fpartheadersize)[0]
820 headersize = self._unpack(_fpartheadersize)[0]
821 if headersize < 0:
821 if headersize < 0:
822 raise error.BundleValueError('negative part header size: %i'
822 raise error.BundleValueError('negative part header size: %i'
823 % headersize)
823 % headersize)
824 self.ui.debug('part header size: %i\n' % headersize)
824 self.ui.debug('part header size: %i\n' % headersize)
825 if headersize:
825 if headersize:
826 return self._readexact(headersize)
826 return self._readexact(headersize)
827 return None
827 return None
828
828
829 def __call__(self):
829 def __call__(self):
830 self.ui.debug('bundle2 stream interruption, looking for a part.\n')
830 self.ui.debug('bundle2 stream interruption, looking for a part.\n')
831 headerblock = self._readpartheader()
831 headerblock = self._readpartheader()
832 if headerblock is None:
832 if headerblock is None:
833 self.ui.debug('no part found during interruption.\n')
833 self.ui.debug('no part found during interruption.\n')
834 return
834 return
835 part = unbundlepart(self.ui, headerblock, self._fp)
835 part = unbundlepart(self.ui, headerblock, self._fp)
836 op = interruptoperation(self.ui)
836 op = interruptoperation(self.ui)
837 _processpart(op, part)
837 _processpart(op, part)
838
838
839 class interruptoperation(object):
839 class interruptoperation(object):
840 """A limited operation to be use by part handler during interruption
840 """A limited operation to be use by part handler during interruption
841
841
842 It only have access to an ui object.
842 It only have access to an ui object.
843 """
843 """
844
844
845 def __init__(self, ui):
845 def __init__(self, ui):
846 self.ui = ui
846 self.ui = ui
847 self.reply = None
847 self.reply = None
848 self.captureoutput = False
848 self.captureoutput = False
849
849
850 @property
850 @property
851 def repo(self):
851 def repo(self):
852 raise RuntimeError('no repo access from stream interruption')
852 raise RuntimeError('no repo access from stream interruption')
853
853
854 def gettransaction(self):
854 def gettransaction(self):
855 raise TransactionUnavailable('no repo access from stream interruption')
855 raise TransactionUnavailable('no repo access from stream interruption')
856
856
857 class unbundlepart(unpackermixin):
857 class unbundlepart(unpackermixin):
858 """a bundle part read from a bundle"""
858 """a bundle part read from a bundle"""
859
859
860 def __init__(self, ui, header, fp):
860 def __init__(self, ui, header, fp):
861 super(unbundlepart, self).__init__(fp)
861 super(unbundlepart, self).__init__(fp)
862 self.ui = ui
862 self.ui = ui
863 # unbundle state attr
863 # unbundle state attr
864 self._headerdata = header
864 self._headerdata = header
865 self._headeroffset = 0
865 self._headeroffset = 0
866 self._initialized = False
866 self._initialized = False
867 self.consumed = False
867 self.consumed = False
868 # part data
868 # part data
869 self.id = None
869 self.id = None
870 self.type = None
870 self.type = None
871 self.mandatoryparams = None
871 self.mandatoryparams = None
872 self.advisoryparams = None
872 self.advisoryparams = None
873 self.params = None
873 self.params = None
874 self.mandatorykeys = ()
874 self.mandatorykeys = ()
875 self._payloadstream = None
875 self._payloadstream = None
876 self._readheader()
876 self._readheader()
877 self._mandatory = None
877 self._mandatory = None
878 self._chunkindex = [] #(payload, file) position tuples for chunk starts
878 self._chunkindex = [] #(payload, file) position tuples for chunk starts
879 self._pos = 0
879 self._pos = 0
880
880
881 def _fromheader(self, size):
881 def _fromheader(self, size):
882 """return the next <size> byte from the header"""
882 """return the next <size> byte from the header"""
883 offset = self._headeroffset
883 offset = self._headeroffset
884 data = self._headerdata[offset:(offset + size)]
884 data = self._headerdata[offset:(offset + size)]
885 self._headeroffset = offset + size
885 self._headeroffset = offset + size
886 return data
886 return data
887
887
888 def _unpackheader(self, format):
888 def _unpackheader(self, format):
889 """read given format from header
889 """read given format from header
890
890
891 This automatically compute the size of the format to read."""
891 This automatically compute the size of the format to read."""
892 data = self._fromheader(struct.calcsize(format))
892 data = self._fromheader(struct.calcsize(format))
893 return _unpack(format, data)
893 return _unpack(format, data)
894
894
895 def _initparams(self, mandatoryparams, advisoryparams):
895 def _initparams(self, mandatoryparams, advisoryparams):
896 """internal function to setup all logic related parameters"""
896 """internal function to setup all logic related parameters"""
897 # make it read only to prevent people touching it by mistake.
897 # make it read only to prevent people touching it by mistake.
898 self.mandatoryparams = tuple(mandatoryparams)
898 self.mandatoryparams = tuple(mandatoryparams)
899 self.advisoryparams = tuple(advisoryparams)
899 self.advisoryparams = tuple(advisoryparams)
900 # user friendly UI
900 # user friendly UI
901 self.params = dict(self.mandatoryparams)
901 self.params = dict(self.mandatoryparams)
902 self.params.update(dict(self.advisoryparams))
902 self.params.update(dict(self.advisoryparams))
903 self.mandatorykeys = frozenset(p[0] for p in mandatoryparams)
903 self.mandatorykeys = frozenset(p[0] for p in mandatoryparams)
904
904
905 def _payloadchunks(self, chunknum=0):
905 def _payloadchunks(self, chunknum=0):
906 '''seek to specified chunk and start yielding data'''
906 '''seek to specified chunk and start yielding data'''
907 if len(self._chunkindex) == 0:
907 if len(self._chunkindex) == 0:
908 assert chunknum == 0, 'Must start with chunk 0'
908 assert chunknum == 0, 'Must start with chunk 0'
909 self._chunkindex.append((0, super(unbundlepart, self).tell()))
909 self._chunkindex.append((0, super(unbundlepart, self).tell()))
910 else:
910 else:
911 assert chunknum < len(self._chunkindex), \
911 assert chunknum < len(self._chunkindex), \
912 'Unknown chunk %d' % chunknum
912 'Unknown chunk %d' % chunknum
913 super(unbundlepart, self).seek(self._chunkindex[chunknum][1])
913 super(unbundlepart, self).seek(self._chunkindex[chunknum][1])
914
914
915 pos = self._chunkindex[chunknum][0]
915 pos = self._chunkindex[chunknum][0]
916 payloadsize = self._unpack(_fpayloadsize)[0]
916 payloadsize = self._unpack(_fpayloadsize)[0]
917 self.ui.debug('payload chunk size: %i\n' % payloadsize)
917 self.ui.debug('payload chunk size: %i\n' % payloadsize)
918 while payloadsize:
918 while payloadsize:
919 if payloadsize == flaginterrupt:
919 if payloadsize == flaginterrupt:
920 # interruption detection, the handler will now read a
920 # interruption detection, the handler will now read a
921 # single part and process it.
921 # single part and process it.
922 interrupthandler(self.ui, self._fp)()
922 interrupthandler(self.ui, self._fp)()
923 elif payloadsize < 0:
923 elif payloadsize < 0:
924 msg = 'negative payload chunk size: %i' % payloadsize
924 msg = 'negative payload chunk size: %i' % payloadsize
925 raise error.BundleValueError(msg)
925 raise error.BundleValueError(msg)
926 else:
926 else:
927 result = self._readexact(payloadsize)
927 result = self._readexact(payloadsize)
928 chunknum += 1
928 chunknum += 1
929 pos += payloadsize
929 pos += payloadsize
930 if chunknum == len(self._chunkindex):
930 if chunknum == len(self._chunkindex):
931 self._chunkindex.append((pos,
931 self._chunkindex.append((pos,
932 super(unbundlepart, self).tell()))
932 super(unbundlepart, self).tell()))
933 yield result
933 yield result
934 payloadsize = self._unpack(_fpayloadsize)[0]
934 payloadsize = self._unpack(_fpayloadsize)[0]
935 self.ui.debug('payload chunk size: %i\n' % payloadsize)
935 self.ui.debug('payload chunk size: %i\n' % payloadsize)
936
936
937 def _findchunk(self, pos):
937 def _findchunk(self, pos):
938 '''for a given payload position, return a chunk number and offset'''
938 '''for a given payload position, return a chunk number and offset'''
939 for chunk, (ppos, fpos) in enumerate(self._chunkindex):
939 for chunk, (ppos, fpos) in enumerate(self._chunkindex):
940 if ppos == pos:
940 if ppos == pos:
941 return chunk, 0
941 return chunk, 0
942 elif ppos > pos:
942 elif ppos > pos:
943 return chunk - 1, pos - self._chunkindex[chunk - 1][0]
943 return chunk - 1, pos - self._chunkindex[chunk - 1][0]
944 raise ValueError('Unknown chunk')
944 raise ValueError('Unknown chunk')
945
945
946 def _readheader(self):
946 def _readheader(self):
947 """read the header and setup the object"""
947 """read the header and setup the object"""
948 typesize = self._unpackheader(_fparttypesize)[0]
948 typesize = self._unpackheader(_fparttypesize)[0]
949 self.type = self._fromheader(typesize)
949 self.type = self._fromheader(typesize)
950 self.ui.debug('part type: "%s"\n' % self.type)
950 self.ui.debug('part type: "%s"\n' % self.type)
951 self.id = self._unpackheader(_fpartid)[0]
951 self.id = self._unpackheader(_fpartid)[0]
952 self.ui.debug('part id: "%s"\n' % self.id)
952 self.ui.debug('part id: "%s"\n' % self.id)
953 # extract mandatory bit from type
953 # extract mandatory bit from type
954 self.mandatory = (self.type != self.type.lower())
954 self.mandatory = (self.type != self.type.lower())
955 self.type = self.type.lower()
955 self.type = self.type.lower()
956 ## reading parameters
956 ## reading parameters
957 # param count
957 # param count
958 mancount, advcount = self._unpackheader(_fpartparamcount)
958 mancount, advcount = self._unpackheader(_fpartparamcount)
959 self.ui.debug('part parameters: %i\n' % (mancount + advcount))
959 self.ui.debug('part parameters: %i\n' % (mancount + advcount))
960 # param size
960 # param size
961 fparamsizes = _makefpartparamsizes(mancount + advcount)
961 fparamsizes = _makefpartparamsizes(mancount + advcount)
962 paramsizes = self._unpackheader(fparamsizes)
962 paramsizes = self._unpackheader(fparamsizes)
963 # make it a list of couple again
963 # make it a list of couple again
964 paramsizes = zip(paramsizes[::2], paramsizes[1::2])
964 paramsizes = zip(paramsizes[::2], paramsizes[1::2])
965 # split mandatory from advisory
965 # split mandatory from advisory
966 mansizes = paramsizes[:mancount]
966 mansizes = paramsizes[:mancount]
967 advsizes = paramsizes[mancount:]
967 advsizes = paramsizes[mancount:]
968 # retrieve param value
968 # retrieve param value
969 manparams = []
969 manparams = []
970 for key, value in mansizes:
970 for key, value in mansizes:
971 manparams.append((self._fromheader(key), self._fromheader(value)))
971 manparams.append((self._fromheader(key), self._fromheader(value)))
972 advparams = []
972 advparams = []
973 for key, value in advsizes:
973 for key, value in advsizes:
974 advparams.append((self._fromheader(key), self._fromheader(value)))
974 advparams.append((self._fromheader(key), self._fromheader(value)))
975 self._initparams(manparams, advparams)
975 self._initparams(manparams, advparams)
976 ## part payload
976 ## part payload
977 self._payloadstream = util.chunkbuffer(self._payloadchunks())
977 self._payloadstream = util.chunkbuffer(self._payloadchunks())
978 # we read the data, tell it
978 # we read the data, tell it
979 self._initialized = True
979 self._initialized = True
980
980
981 def read(self, size=None):
981 def read(self, size=None):
982 """read payload data"""
982 """read payload data"""
983 if not self._initialized:
983 if not self._initialized:
984 self._readheader()
984 self._readheader()
985 if size is None:
985 if size is None:
986 data = self._payloadstream.read()
986 data = self._payloadstream.read()
987 else:
987 else:
988 data = self._payloadstream.read(size)
988 data = self._payloadstream.read(size)
989 if size is None or len(data) < size:
989 if size is None or len(data) < size:
990 self.consumed = True
990 self.consumed = True
991 self._pos += len(data)
991 self._pos += len(data)
992 return data
992 return data
993
993
994 def tell(self):
994 def tell(self):
995 return self._pos
995 return self._pos
996
996
997 def seek(self, offset, whence=0):
997 def seek(self, offset, whence=0):
998 if whence == 0:
998 if whence == 0:
999 newpos = offset
999 newpos = offset
1000 elif whence == 1:
1000 elif whence == 1:
1001 newpos = self._pos + offset
1001 newpos = self._pos + offset
1002 elif whence == 2:
1002 elif whence == 2:
1003 if not self.consumed:
1003 if not self.consumed:
1004 self.read()
1004 self.read()
1005 newpos = self._chunkindex[-1][0] - offset
1005 newpos = self._chunkindex[-1][0] - offset
1006 else:
1006 else:
1007 raise ValueError('Unknown whence value: %r' % (whence,))
1007 raise ValueError('Unknown whence value: %r' % (whence,))
1008
1008
1009 if newpos > self._chunkindex[-1][0] and not self.consumed:
1009 if newpos > self._chunkindex[-1][0] and not self.consumed:
1010 self.read()
1010 self.read()
1011 if not 0 <= newpos <= self._chunkindex[-1][0]:
1011 if not 0 <= newpos <= self._chunkindex[-1][0]:
1012 raise ValueError('Offset out of range')
1012 raise ValueError('Offset out of range')
1013
1013
1014 if self._pos != newpos:
1014 if self._pos != newpos:
1015 chunk, internaloffset = self._findchunk(newpos)
1015 chunk, internaloffset = self._findchunk(newpos)
1016 self._payloadstream = util.chunkbuffer(self._payloadchunks(chunk))
1016 self._payloadstream = util.chunkbuffer(self._payloadchunks(chunk))
1017 adjust = self.read(internaloffset)
1017 adjust = self.read(internaloffset)
1018 if len(adjust) != internaloffset:
1018 if len(adjust) != internaloffset:
1019 raise util.Abort(_('Seek failed\n'))
1019 raise util.Abort(_('Seek failed\n'))
1020 self._pos = newpos
1020 self._pos = newpos
1021
1021
1022 # These are only the static capabilities.
1023 # Check the 'getrepocaps' function for the rest.
1022 capabilities = {'HG20': (),
1024 capabilities = {'HG20': (),
1023 'listkeys': (),
1025 'listkeys': (),
1024 'pushkey': (),
1026 'pushkey': (),
1025 'digests': tuple(sorted(util.DIGESTS.keys())),
1027 'digests': tuple(sorted(util.DIGESTS.keys())),
1026 'remote-changegroup': ('http', 'https'),
1028 'remote-changegroup': ('http', 'https'),
1027 }
1029 }
1028
1030
1029 def getrepocaps(repo, allowpushback=False):
1031 def getrepocaps(repo, allowpushback=False):
1030 """return the bundle2 capabilities for a given repo
1032 """return the bundle2 capabilities for a given repo
1031
1033
1032 Exists to allow extensions (like evolution) to mutate the capabilities.
1034 Exists to allow extensions (like evolution) to mutate the capabilities.
1033 """
1035 """
1034 caps = capabilities.copy()
1036 caps = capabilities.copy()
1035 caps['changegroup'] = tuple(sorted(changegroup.packermap.keys()))
1037 caps['changegroup'] = tuple(sorted(changegroup.packermap.keys()))
1036 if obsolete.isenabled(repo, obsolete.exchangeopt):
1038 if obsolete.isenabled(repo, obsolete.exchangeopt):
1037 supportedformat = tuple('V%i' % v for v in obsolete.formats)
1039 supportedformat = tuple('V%i' % v for v in obsolete.formats)
1038 caps['obsmarkers'] = supportedformat
1040 caps['obsmarkers'] = supportedformat
1039 if allowpushback:
1041 if allowpushback:
1040 caps['pushback'] = ()
1042 caps['pushback'] = ()
1041 return caps
1043 return caps
1042
1044
1043 def bundle2caps(remote):
1045 def bundle2caps(remote):
1044 """return the bundle capabilities of a peer as dict"""
1046 """return the bundle capabilities of a peer as dict"""
1045 raw = remote.capable('bundle2')
1047 raw = remote.capable('bundle2')
1046 if not raw and raw != '':
1048 if not raw and raw != '':
1047 return {}
1049 return {}
1048 capsblob = urllib.unquote(remote.capable('bundle2'))
1050 capsblob = urllib.unquote(remote.capable('bundle2'))
1049 return decodecaps(capsblob)
1051 return decodecaps(capsblob)
1050
1052
1051 def obsmarkersversion(caps):
1053 def obsmarkersversion(caps):
1052 """extract the list of supported obsmarkers versions from a bundle2caps dict
1054 """extract the list of supported obsmarkers versions from a bundle2caps dict
1053 """
1055 """
1054 obscaps = caps.get('obsmarkers', ())
1056 obscaps = caps.get('obsmarkers', ())
1055 return [int(c[1:]) for c in obscaps if c.startswith('V')]
1057 return [int(c[1:]) for c in obscaps if c.startswith('V')]
1056
1058
1057 @parthandler('changegroup', ('version',))
1059 @parthandler('changegroup', ('version',))
1058 def handlechangegroup(op, inpart):
1060 def handlechangegroup(op, inpart):
1059 """apply a changegroup part on the repo
1061 """apply a changegroup part on the repo
1060
1062
1061 This is a very early implementation that will massive rework before being
1063 This is a very early implementation that will massive rework before being
1062 inflicted to any end-user.
1064 inflicted to any end-user.
1063 """
1065 """
1064 # Make sure we trigger a transaction creation
1066 # Make sure we trigger a transaction creation
1065 #
1067 #
1066 # The addchangegroup function will get a transaction object by itself, but
1068 # The addchangegroup function will get a transaction object by itself, but
1067 # we need to make sure we trigger the creation of a transaction object used
1069 # we need to make sure we trigger the creation of a transaction object used
1068 # for the whole processing scope.
1070 # for the whole processing scope.
1069 op.gettransaction()
1071 op.gettransaction()
1070 unpackerversion = inpart.params.get('version', '01')
1072 unpackerversion = inpart.params.get('version', '01')
1071 # We should raise an appropriate exception here
1073 # We should raise an appropriate exception here
1072 unpacker = changegroup.packermap[unpackerversion][1]
1074 unpacker = changegroup.packermap[unpackerversion][1]
1073 cg = unpacker(inpart, 'UN')
1075 cg = unpacker(inpart, 'UN')
1074 # the source and url passed here are overwritten by the one contained in
1076 # the source and url passed here are overwritten by the one contained in
1075 # the transaction.hookargs argument. So 'bundle2' is a placeholder
1077 # the transaction.hookargs argument. So 'bundle2' is a placeholder
1076 ret = changegroup.addchangegroup(op.repo, cg, 'bundle2', 'bundle2')
1078 ret = changegroup.addchangegroup(op.repo, cg, 'bundle2', 'bundle2')
1077 op.records.add('changegroup', {'return': ret})
1079 op.records.add('changegroup', {'return': ret})
1078 if op.reply is not None:
1080 if op.reply is not None:
1079 # This is definitely not the final form of this
1081 # This is definitely not the final form of this
1080 # return. But one need to start somewhere.
1082 # return. But one need to start somewhere.
1081 part = op.reply.newpart('reply:changegroup', mandatory=False)
1083 part = op.reply.newpart('reply:changegroup', mandatory=False)
1082 part.addparam('in-reply-to', str(inpart.id), mandatory=False)
1084 part.addparam('in-reply-to', str(inpart.id), mandatory=False)
1083 part.addparam('return', '%i' % ret, mandatory=False)
1085 part.addparam('return', '%i' % ret, mandatory=False)
1084 assert not inpart.read()
1086 assert not inpart.read()
1085
1087
1086 _remotechangegroupparams = tuple(['url', 'size', 'digests'] +
1088 _remotechangegroupparams = tuple(['url', 'size', 'digests'] +
1087 ['digest:%s' % k for k in util.DIGESTS.keys()])
1089 ['digest:%s' % k for k in util.DIGESTS.keys()])
1088 @parthandler('remote-changegroup', _remotechangegroupparams)
1090 @parthandler('remote-changegroup', _remotechangegroupparams)
1089 def handleremotechangegroup(op, inpart):
1091 def handleremotechangegroup(op, inpart):
1090 """apply a bundle10 on the repo, given an url and validation information
1092 """apply a bundle10 on the repo, given an url and validation information
1091
1093
1092 All the information about the remote bundle to import are given as
1094 All the information about the remote bundle to import are given as
1093 parameters. The parameters include:
1095 parameters. The parameters include:
1094 - url: the url to the bundle10.
1096 - url: the url to the bundle10.
1095 - size: the bundle10 file size. It is used to validate what was
1097 - size: the bundle10 file size. It is used to validate what was
1096 retrieved by the client matches the server knowledge about the bundle.
1098 retrieved by the client matches the server knowledge about the bundle.
1097 - digests: a space separated list of the digest types provided as
1099 - digests: a space separated list of the digest types provided as
1098 parameters.
1100 parameters.
1099 - digest:<digest-type>: the hexadecimal representation of the digest with
1101 - digest:<digest-type>: the hexadecimal representation of the digest with
1100 that name. Like the size, it is used to validate what was retrieved by
1102 that name. Like the size, it is used to validate what was retrieved by
1101 the client matches what the server knows about the bundle.
1103 the client matches what the server knows about the bundle.
1102
1104
1103 When multiple digest types are given, all of them are checked.
1105 When multiple digest types are given, all of them are checked.
1104 """
1106 """
1105 try:
1107 try:
1106 raw_url = inpart.params['url']
1108 raw_url = inpart.params['url']
1107 except KeyError:
1109 except KeyError:
1108 raise util.Abort(_('remote-changegroup: missing "%s" param') % 'url')
1110 raise util.Abort(_('remote-changegroup: missing "%s" param') % 'url')
1109 parsed_url = util.url(raw_url)
1111 parsed_url = util.url(raw_url)
1110 if parsed_url.scheme not in capabilities['remote-changegroup']:
1112 if parsed_url.scheme not in capabilities['remote-changegroup']:
1111 raise util.Abort(_('remote-changegroup does not support %s urls') %
1113 raise util.Abort(_('remote-changegroup does not support %s urls') %
1112 parsed_url.scheme)
1114 parsed_url.scheme)
1113
1115
1114 try:
1116 try:
1115 size = int(inpart.params['size'])
1117 size = int(inpart.params['size'])
1116 except ValueError:
1118 except ValueError:
1117 raise util.Abort(_('remote-changegroup: invalid value for param "%s"')
1119 raise util.Abort(_('remote-changegroup: invalid value for param "%s"')
1118 % 'size')
1120 % 'size')
1119 except KeyError:
1121 except KeyError:
1120 raise util.Abort(_('remote-changegroup: missing "%s" param') % 'size')
1122 raise util.Abort(_('remote-changegroup: missing "%s" param') % 'size')
1121
1123
1122 digests = {}
1124 digests = {}
1123 for typ in inpart.params.get('digests', '').split():
1125 for typ in inpart.params.get('digests', '').split():
1124 param = 'digest:%s' % typ
1126 param = 'digest:%s' % typ
1125 try:
1127 try:
1126 value = inpart.params[param]
1128 value = inpart.params[param]
1127 except KeyError:
1129 except KeyError:
1128 raise util.Abort(_('remote-changegroup: missing "%s" param') %
1130 raise util.Abort(_('remote-changegroup: missing "%s" param') %
1129 param)
1131 param)
1130 digests[typ] = value
1132 digests[typ] = value
1131
1133
1132 real_part = util.digestchecker(url.open(op.ui, raw_url), size, digests)
1134 real_part = util.digestchecker(url.open(op.ui, raw_url), size, digests)
1133
1135
1134 # Make sure we trigger a transaction creation
1136 # Make sure we trigger a transaction creation
1135 #
1137 #
1136 # The addchangegroup function will get a transaction object by itself, but
1138 # The addchangegroup function will get a transaction object by itself, but
1137 # we need to make sure we trigger the creation of a transaction object used
1139 # we need to make sure we trigger the creation of a transaction object used
1138 # for the whole processing scope.
1140 # for the whole processing scope.
1139 op.gettransaction()
1141 op.gettransaction()
1140 import exchange
1142 import exchange
1141 cg = exchange.readbundle(op.repo.ui, real_part, raw_url)
1143 cg = exchange.readbundle(op.repo.ui, real_part, raw_url)
1142 if not isinstance(cg, changegroup.cg1unpacker):
1144 if not isinstance(cg, changegroup.cg1unpacker):
1143 raise util.Abort(_('%s: not a bundle version 1.0') %
1145 raise util.Abort(_('%s: not a bundle version 1.0') %
1144 util.hidepassword(raw_url))
1146 util.hidepassword(raw_url))
1145 ret = changegroup.addchangegroup(op.repo, cg, 'bundle2', 'bundle2')
1147 ret = changegroup.addchangegroup(op.repo, cg, 'bundle2', 'bundle2')
1146 op.records.add('changegroup', {'return': ret})
1148 op.records.add('changegroup', {'return': ret})
1147 if op.reply is not None:
1149 if op.reply is not None:
1148 # This is definitely not the final form of this
1150 # This is definitely not the final form of this
1149 # return. But one need to start somewhere.
1151 # return. But one need to start somewhere.
1150 part = op.reply.newpart('reply:changegroup')
1152 part = op.reply.newpart('reply:changegroup')
1151 part.addparam('in-reply-to', str(inpart.id), mandatory=False)
1153 part.addparam('in-reply-to', str(inpart.id), mandatory=False)
1152 part.addparam('return', '%i' % ret, mandatory=False)
1154 part.addparam('return', '%i' % ret, mandatory=False)
1153 try:
1155 try:
1154 real_part.validate()
1156 real_part.validate()
1155 except util.Abort, e:
1157 except util.Abort, e:
1156 raise util.Abort(_('bundle at %s is corrupted:\n%s') %
1158 raise util.Abort(_('bundle at %s is corrupted:\n%s') %
1157 (util.hidepassword(raw_url), str(e)))
1159 (util.hidepassword(raw_url), str(e)))
1158 assert not inpart.read()
1160 assert not inpart.read()
1159
1161
1160 @parthandler('reply:changegroup', ('return', 'in-reply-to'))
1162 @parthandler('reply:changegroup', ('return', 'in-reply-to'))
1161 def handlereplychangegroup(op, inpart):
1163 def handlereplychangegroup(op, inpart):
1162 ret = int(inpart.params['return'])
1164 ret = int(inpart.params['return'])
1163 replyto = int(inpart.params['in-reply-to'])
1165 replyto = int(inpart.params['in-reply-to'])
1164 op.records.add('changegroup', {'return': ret}, replyto)
1166 op.records.add('changegroup', {'return': ret}, replyto)
1165
1167
1166 @parthandler('check:heads')
1168 @parthandler('check:heads')
1167 def handlecheckheads(op, inpart):
1169 def handlecheckheads(op, inpart):
1168 """check that head of the repo did not change
1170 """check that head of the repo did not change
1169
1171
1170 This is used to detect a push race when using unbundle.
1172 This is used to detect a push race when using unbundle.
1171 This replaces the "heads" argument of unbundle."""
1173 This replaces the "heads" argument of unbundle."""
1172 h = inpart.read(20)
1174 h = inpart.read(20)
1173 heads = []
1175 heads = []
1174 while len(h) == 20:
1176 while len(h) == 20:
1175 heads.append(h)
1177 heads.append(h)
1176 h = inpart.read(20)
1178 h = inpart.read(20)
1177 assert not h
1179 assert not h
1178 if heads != op.repo.heads():
1180 if heads != op.repo.heads():
1179 raise error.PushRaced('repository changed while pushing - '
1181 raise error.PushRaced('repository changed while pushing - '
1180 'please try again')
1182 'please try again')
1181
1183
1182 @parthandler('output')
1184 @parthandler('output')
1183 def handleoutput(op, inpart):
1185 def handleoutput(op, inpart):
1184 """forward output captured on the server to the client"""
1186 """forward output captured on the server to the client"""
1185 for line in inpart.read().splitlines():
1187 for line in inpart.read().splitlines():
1186 op.ui.status(('remote: %s\n' % line))
1188 op.ui.status(('remote: %s\n' % line))
1187
1189
1188 @parthandler('replycaps')
1190 @parthandler('replycaps')
1189 def handlereplycaps(op, inpart):
1191 def handlereplycaps(op, inpart):
1190 """Notify that a reply bundle should be created
1192 """Notify that a reply bundle should be created
1191
1193
1192 The payload contains the capabilities information for the reply"""
1194 The payload contains the capabilities information for the reply"""
1193 caps = decodecaps(inpart.read())
1195 caps = decodecaps(inpart.read())
1194 if op.reply is None:
1196 if op.reply is None:
1195 op.reply = bundle20(op.ui, caps)
1197 op.reply = bundle20(op.ui, caps)
1196
1198
1197 @parthandler('error:abort', ('message', 'hint'))
1199 @parthandler('error:abort', ('message', 'hint'))
1198 def handleerrorabort(op, inpart):
1200 def handleerrorabort(op, inpart):
1199 """Used to transmit abort error over the wire"""
1201 """Used to transmit abort error over the wire"""
1200 raise util.Abort(inpart.params['message'], hint=inpart.params.get('hint'))
1202 raise util.Abort(inpart.params['message'], hint=inpart.params.get('hint'))
1201
1203
1202 @parthandler('error:unsupportedcontent', ('parttype', 'params'))
1204 @parthandler('error:unsupportedcontent', ('parttype', 'params'))
1203 def handleerrorunsupportedcontent(op, inpart):
1205 def handleerrorunsupportedcontent(op, inpart):
1204 """Used to transmit unknown content error over the wire"""
1206 """Used to transmit unknown content error over the wire"""
1205 kwargs = {}
1207 kwargs = {}
1206 parttype = inpart.params.get('parttype')
1208 parttype = inpart.params.get('parttype')
1207 if parttype is not None:
1209 if parttype is not None:
1208 kwargs['parttype'] = parttype
1210 kwargs['parttype'] = parttype
1209 params = inpart.params.get('params')
1211 params = inpart.params.get('params')
1210 if params is not None:
1212 if params is not None:
1211 kwargs['params'] = params.split('\0')
1213 kwargs['params'] = params.split('\0')
1212
1214
1213 raise error.UnsupportedPartError(**kwargs)
1215 raise error.UnsupportedPartError(**kwargs)
1214
1216
1215 @parthandler('error:pushraced', ('message',))
1217 @parthandler('error:pushraced', ('message',))
1216 def handleerrorpushraced(op, inpart):
1218 def handleerrorpushraced(op, inpart):
1217 """Used to transmit push race error over the wire"""
1219 """Used to transmit push race error over the wire"""
1218 raise error.ResponseError(_('push failed:'), inpart.params['message'])
1220 raise error.ResponseError(_('push failed:'), inpart.params['message'])
1219
1221
1220 @parthandler('listkeys', ('namespace',))
1222 @parthandler('listkeys', ('namespace',))
1221 def handlelistkeys(op, inpart):
1223 def handlelistkeys(op, inpart):
1222 """retrieve pushkey namespace content stored in a bundle2"""
1224 """retrieve pushkey namespace content stored in a bundle2"""
1223 namespace = inpart.params['namespace']
1225 namespace = inpart.params['namespace']
1224 r = pushkey.decodekeys(inpart.read())
1226 r = pushkey.decodekeys(inpart.read())
1225 op.records.add('listkeys', (namespace, r))
1227 op.records.add('listkeys', (namespace, r))
1226
1228
1227 @parthandler('pushkey', ('namespace', 'key', 'old', 'new'))
1229 @parthandler('pushkey', ('namespace', 'key', 'old', 'new'))
1228 def handlepushkey(op, inpart):
1230 def handlepushkey(op, inpart):
1229 """process a pushkey request"""
1231 """process a pushkey request"""
1230 dec = pushkey.decode
1232 dec = pushkey.decode
1231 namespace = dec(inpart.params['namespace'])
1233 namespace = dec(inpart.params['namespace'])
1232 key = dec(inpart.params['key'])
1234 key = dec(inpart.params['key'])
1233 old = dec(inpart.params['old'])
1235 old = dec(inpart.params['old'])
1234 new = dec(inpart.params['new'])
1236 new = dec(inpart.params['new'])
1235 ret = op.repo.pushkey(namespace, key, old, new)
1237 ret = op.repo.pushkey(namespace, key, old, new)
1236 record = {'namespace': namespace,
1238 record = {'namespace': namespace,
1237 'key': key,
1239 'key': key,
1238 'old': old,
1240 'old': old,
1239 'new': new}
1241 'new': new}
1240 op.records.add('pushkey', record)
1242 op.records.add('pushkey', record)
1241 if op.reply is not None:
1243 if op.reply is not None:
1242 rpart = op.reply.newpart('reply:pushkey')
1244 rpart = op.reply.newpart('reply:pushkey')
1243 rpart.addparam('in-reply-to', str(inpart.id), mandatory=False)
1245 rpart.addparam('in-reply-to', str(inpart.id), mandatory=False)
1244 rpart.addparam('return', '%i' % ret, mandatory=False)
1246 rpart.addparam('return', '%i' % ret, mandatory=False)
1245
1247
1246 @parthandler('reply:pushkey', ('return', 'in-reply-to'))
1248 @parthandler('reply:pushkey', ('return', 'in-reply-to'))
1247 def handlepushkeyreply(op, inpart):
1249 def handlepushkeyreply(op, inpart):
1248 """retrieve the result of a pushkey request"""
1250 """retrieve the result of a pushkey request"""
1249 ret = int(inpart.params['return'])
1251 ret = int(inpart.params['return'])
1250 partid = int(inpart.params['in-reply-to'])
1252 partid = int(inpart.params['in-reply-to'])
1251 op.records.add('pushkey', {'return': ret}, partid)
1253 op.records.add('pushkey', {'return': ret}, partid)
1252
1254
1253 @parthandler('obsmarkers')
1255 @parthandler('obsmarkers')
1254 def handleobsmarker(op, inpart):
1256 def handleobsmarker(op, inpart):
1255 """add a stream of obsmarkers to the repo"""
1257 """add a stream of obsmarkers to the repo"""
1256 tr = op.gettransaction()
1258 tr = op.gettransaction()
1257 markerdata = inpart.read()
1259 markerdata = inpart.read()
1258 if op.ui.config('experimental', 'obsmarkers-exchange-debug', False):
1260 if op.ui.config('experimental', 'obsmarkers-exchange-debug', False):
1259 op.ui.write(('obsmarker-exchange: %i bytes received\n')
1261 op.ui.write(('obsmarker-exchange: %i bytes received\n')
1260 % len(markerdata))
1262 % len(markerdata))
1261 new = op.repo.obsstore.mergemarkers(tr, markerdata)
1263 new = op.repo.obsstore.mergemarkers(tr, markerdata)
1262 if new:
1264 if new:
1263 op.repo.ui.status(_('%i new obsolescence markers\n') % new)
1265 op.repo.ui.status(_('%i new obsolescence markers\n') % new)
1264 op.records.add('obsmarkers', {'new': new})
1266 op.records.add('obsmarkers', {'new': new})
1265 if op.reply is not None:
1267 if op.reply is not None:
1266 rpart = op.reply.newpart('reply:obsmarkers')
1268 rpart = op.reply.newpart('reply:obsmarkers')
1267 rpart.addparam('in-reply-to', str(inpart.id), mandatory=False)
1269 rpart.addparam('in-reply-to', str(inpart.id), mandatory=False)
1268 rpart.addparam('new', '%i' % new, mandatory=False)
1270 rpart.addparam('new', '%i' % new, mandatory=False)
1269
1271
1270
1272
1271 @parthandler('reply:obsmarkers', ('new', 'in-reply-to'))
1273 @parthandler('reply:obsmarkers', ('new', 'in-reply-to'))
1272 def handlepushkeyreply(op, inpart):
1274 def handlepushkeyreply(op, inpart):
1273 """retrieve the result of a pushkey request"""
1275 """retrieve the result of a pushkey request"""
1274 ret = int(inpart.params['new'])
1276 ret = int(inpart.params['new'])
1275 partid = int(inpart.params['in-reply-to'])
1277 partid = int(inpart.params['in-reply-to'])
1276 op.records.add('obsmarkers', {'new': ret}, partid)
1278 op.records.add('obsmarkers', {'new': ret}, partid)
General Comments 0
You need to be logged in to leave comments. Login now