upstream/mercurial-mirror Commit - r38723:f016eac2

1

# bundle2.py - generic container format to transmit arbitrary data.

1

# bundle2.py - generic container format to transmit arbitrary data.

2

#

2

#

3

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

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

13

14

General format architecture

14

General format architecture

15

===========================

15

===========================

16

17

The format is architectured as follow

17

The format is architectured as follow

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

24

the Binary format

24

the Binary format

25

============================

25

============================

26

27

All numbers are unsigned and big-endian.

27

All numbers are unsigned and big-endian.

28

29

stream level parameters

29

stream level parameters

30

------------------------

30

------------------------

31

32

Binary format is as follow

32

Binary format is as follow

33

34

:params size: int32

34

:params size: int32

35

36

The total number of Bytes used by the parameters

36

The total number of Bytes used by the parameters

37

38

:params value: arbitrary number of Bytes

38

:params value: arbitrary number of Bytes

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

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

46

Empty name are obviously forbidden.

46

Empty name are obviously forbidden.

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

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

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

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

62

Payload part

62

Payload part

63

------------------------

63

------------------------

64

65

Binary format is as follow

65

Binary format is as follow

66

67

:header size: int32

67

:header size: int32

68

69

The total number of Bytes used by the part header. When the header is empty

69

The total number of Bytes used by the part header. 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

72

:header:

72

:header:

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

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

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

84

The binary format of the header is has follow

84

The binary format of the header is has follow

85

86

:typesize: (one byte)

86

:typesize: (one byte)

87

88

:parttype: alphanumerical part name (restricted to [a-zA-Z0-9_:-]*)

88

:parttype: alphanumerical part name (restricted to [a-zA-Z0-9_:-]*)

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

93

:parameters:

93

:parameters:

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

97

<mandatory-count><advisory-count><param-sizes><param-data>

97

<mandatory-count><advisory-count><param-sizes><param-data>

98

99

:mandatory-count: 1 byte, number of mandatory parameters

99

:mandatory-count: 1 byte, number of mandatory parameters

100

101

:advisory-count: 1 byte, number of advisory parameters

101

:advisory-count: 1 byte, number of advisory parameters

102

103

:param-sizes:

103

:param-sizes:

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

108

:param-data:

108

:param-data:

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

114

Mandatory parameters comes first, then the advisory ones.

114

Mandatory parameters comes first, then the advisory ones.

115

116

Each parameter's key MUST be unique within the part.

116

Each parameter's key MUST be unique within the part.

117

118

:payload:

118

:payload:

119

120

payload is a series of `<chunksize><chunkdata>`.

120

payload is a series of `<chunksize><chunkdata>`.

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

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

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

131

Bundle processing

131

Bundle processing

132

============================

132

============================

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

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

148

from __future__ import absolute_import, division

148

from __future__ import absolute_import, division

149

150

import collections

150

import collections

151

import errno

151

import errno

152

import os

152

import os

153

import re

153

import re

154

import string

154

import string

155

import struct

155

import struct

156

import sys

156

import sys

157

158

from .i18n import _

158

from .i18n import _

159

from . import (

159

from . import (

160

bookmarks,

160

bookmarks,

161

changegroup,

161

changegroup,

162

encoding,

162

encoding,

163

error,

163

error,

164

node as nodemod,

164

node as nodemod,

165

obsolete,

165

obsolete,

166

phases,

166

phases,

167

pushkey,

167

pushkey,

168

pycompat,

168

pycompat,

169

streamclone,

169

streamclone,

170

tags,

170

tags,

171

url,

171

url,

172

util,

172

util,

173

)

173

)

174

from .utils import (

174

from .utils import (

175

stringutil,

175

stringutil,

176

)

176

)

177

178

urlerr = util.urlerr

178

urlerr = util.urlerr

179

urlreq = util.urlreq

179

urlreq = util.urlreq

180

181

_pack = struct.pack

181

_pack = struct.pack

182

_unpack = struct.unpack

182

_unpack = struct.unpack

183

184

_fstreamparamsize = '>i'

184

_fstreamparamsize = '>i'

185

_fpartheadersize = '>i'

185

_fpartheadersize = '>i'

186

_fparttypesize = '>B'

186

_fparttypesize = '>B'

187

_fpartid = '>I'

187

_fpartid = '>I'

188

_fpayloadsize = '>i'

188

_fpayloadsize = '>i'

189

_fpartparamcount = '>BB'

189

_fpartparamcount = '>BB'

190

191

preferedchunksize = 32768

191

preferedchunksize = 32768

192

193

_parttypeforbidden = re.compile('[^a-zA-Z0-9_:-]')

193

_parttypeforbidden = re.compile('[^a-zA-Z0-9_:-]')

194

195

def outdebug(ui, message):

195

def outdebug(ui, message):

196

"""debug regarding output stream (bundling)"""

196

"""debug regarding output stream (bundling)"""

197

if ui.configbool('devel', 'bundle2.debug'):

197

if ui.configbool('devel', 'bundle2.debug'):

198

ui.debug('bundle2-output: %s\n' % message)

198

ui.debug('bundle2-output: %s\n' % message)

199

200

def indebug(ui, message):

200

def indebug(ui, message):

201

"""debug on input stream (unbundling)"""

201

"""debug on input stream (unbundling)"""

202

if ui.configbool('devel', 'bundle2.debug'):

202

if ui.configbool('devel', 'bundle2.debug'):

203

ui.debug('bundle2-input: %s\n' % message)

203

ui.debug('bundle2-input: %s\n' % message)

204

205

def validateparttype(parttype):

205

def validateparttype(parttype):

206

"""raise ValueError if a parttype contains invalid character"""

206

"""raise ValueError if a parttype contains invalid character"""

207

if _parttypeforbidden.search(parttype):

207

if _parttypeforbidden.search(parttype):

208

raise ValueError(parttype)

208

raise ValueError(parttype)

209

210

def _makefpartparamsizes(nbparams):

210

def _makefpartparamsizes(nbparams):

211

"""return a struct format to read part parameter sizes

211

"""return a struct format to read part parameter sizes

212

213

The number parameters is variable so we need to build that format

213

The number parameters is variable so we need to build that format

214

dynamically.

214

dynamically.

215

"""

215

"""

216

return '>'+('BB'*nbparams)

216

return '>'+('BB'*nbparams)

217

218

parthandlermapping = {}

218

parthandlermapping = {}

219

220

def parthandler(parttype, params=()):

220

def parthandler(parttype, params=()):

221

"""decorator that register a function as a bundle2 part handler

221

"""decorator that register a function as a bundle2 part handler

222

223

eg::

223

eg::

224

225

@parthandler('myparttype', ('mandatory', 'param', 'handled'))

225

@parthandler('myparttype', ('mandatory', 'param', 'handled'))

226

def myparttypehandler(...):

226

def myparttypehandler(...):

227

'''process a part of type "my part".'''

227

'''process a part of type "my part".'''

228

...

228

...

229

"""

229

"""

230

validateparttype(parttype)

230

validateparttype(parttype)

231

def _decorator(func):

231

def _decorator(func):

232

lparttype = parttype.lower() # enforce lower case matching.

232

lparttype = parttype.lower() # enforce lower case matching.

233

assert lparttype not in parthandlermapping

233

assert lparttype not in parthandlermapping

234

parthandlermapping[lparttype] = func

234

parthandlermapping[lparttype] = func

235

func.params = frozenset(params)

235

func.params = frozenset(params)

236

return func

236

return func

237

return _decorator

237

return _decorator

238

239

class unbundlerecords(object):

239

class unbundlerecords(object):

240

"""keep record of what happens during and unbundle

240

"""keep record of what happens during and unbundle

241

242

New records are added using `records.add('cat', obj)`. Where 'cat' is a

242

New records are added using `records.add('cat', obj)`. Where 'cat' is a

243

category of record and obj is an arbitrary object.

243

category of record and obj is an arbitrary object.

244

245

`records['cat']` will return all entries of this category 'cat'.

245

`records['cat']` will return all entries of this category 'cat'.

246

247

Iterating on the object itself will yield `('category', obj)` tuples

247

Iterating on the object itself will yield `('category', obj)` tuples

248

for all entries.

248

for all entries.

249

250

All iterations happens in chronological order.

250

All iterations happens in chronological order.

251

"""

251

"""

252

253

def __init__(self):

253

def __init__(self):

254

self._categories = {}

254

self._categories = {}

255

self._sequences = []

255

self._sequences = []

256

self._replies = {}

256

self._replies = {}

257

258

def add(self, category, entry, inreplyto=None):

258

def add(self, category, entry, inreplyto=None):

259

"""add a new record of a given category.

259

"""add a new record of a given category.

260

261

The entry can then be retrieved in the list returned by

261

The entry can then be retrieved in the list returned by

262

self['category']."""

262

self['category']."""

263

self._categories.setdefault(category, []).append(entry)

263

self._categories.setdefault(category, []).append(entry)

264

self._sequences.append((category, entry))

264

self._sequences.append((category, entry))

265

if inreplyto is not None:

265

if inreplyto is not None:

266

self.getreplies(inreplyto).add(category, entry)

266

self.getreplies(inreplyto).add(category, entry)

267

268

def getreplies(self, partid):

268

def getreplies(self, partid):

269

"""get the records that are replies to a specific part"""

269

"""get the records that are replies to a specific part"""

270

return self._replies.setdefault(partid, unbundlerecords())

270

return self._replies.setdefault(partid, unbundlerecords())

271

272

def __getitem__(self, cat):

272

def __getitem__(self, cat):

273

return tuple(self._categories.get(cat, ()))

273

return tuple(self._categories.get(cat, ()))

274

275

def __iter__(self):

275

def __iter__(self):

276

return iter(self._sequences)

276

return iter(self._sequences)

277

278

def __len__(self):

278

def __len__(self):

279

return len(self._sequences)

279

return len(self._sequences)

280

281

def __nonzero__(self):

281

def __nonzero__(self):

282

return bool(self._sequences)

282

return bool(self._sequences)

283

284

__bool__ = __nonzero__

284

__bool__ = __nonzero__

285

286

class bundleoperation(object):

286

class bundleoperation(object):

287

"""an object that represents a single bundling process

287

"""an object that represents a single bundling process

288

289

Its purpose is to carry unbundle-related objects and states.

289

Its purpose is to carry unbundle-related objects and states.

290

291

A new object should be created at the beginning of each bundle processing.

291

A new object should be created at the beginning of each bundle processing.

292

The object is to be returned by the processing function.

292

The object is to be returned by the processing function.

293

294

The object has very little content now it will ultimately contain:

294

The object has very little content now it will ultimately contain:

295

* an access to the repo the bundle is applied to,

295

* an access to the repo the bundle is applied to,

296

* a ui object,

296

* a ui object,

297

* a way to retrieve a transaction to add changes to the repo,

297

* a way to retrieve a transaction to add changes to the repo,

298

* a way to record the result of processing each part,

298

* a way to record the result of processing each part,

299

* a way to construct a bundle response when applicable.

299

* a way to construct a bundle response when applicable.

300

"""

300

"""

301

302

def __init__(self, repo, transactiongetter, captureoutput=True, source=''):

302

def __init__(self, repo, transactiongetter, captureoutput=True, source=''):

303

self.repo = repo

303

self.repo = repo

304

self.ui = repo.ui

304

self.ui = repo.ui

305

self.records = unbundlerecords()

305

self.records = unbundlerecords()

306

self.reply = None

306

self.reply = None

307

self.captureoutput = captureoutput

307

self.captureoutput = captureoutput

308

self.hookargs = {}

308

self.hookargs = {}

309

self._gettransaction = transactiongetter

309

self._gettransaction = transactiongetter

310

# carries value that can modify part behavior

310

# carries value that can modify part behavior

311

self.modes = {}

311

self.modes = {}

312

self.source = source

312

self.source = source

313

314

def gettransaction(self):

314

def gettransaction(self):

315

transaction = self._gettransaction()

315

transaction = self._gettransaction()

316

317

if self.hookargs:

317

if self.hookargs:

318

# the ones added to the transaction supercede those added

318

# the ones added to the transaction supercede those added

319

# to the operation.

319

# to the operation.

320

self.hookargs.update(transaction.hookargs)

320

self.hookargs.update(transaction.hookargs)

321

transaction.hookargs = self.hookargs

321

transaction.hookargs = self.hookargs

322

323

# mark the hookargs as flushed. further attempts to add to

323

# mark the hookargs as flushed. further attempts to add to

324

# hookargs will result in an abort.

324

# hookargs will result in an abort.

325

self.hookargs = None

325

self.hookargs = None

326

327

return transaction

327

return transaction

328

329

def addhookargs(self, hookargs):

329

def addhookargs(self, hookargs):

330

if self.hookargs is None:

330

if self.hookargs is None:

331

raise error.ProgrammingError('attempted to add hookargs to '

331

raise error.ProgrammingError('attempted to add hookargs to '

332

'operation after transaction started')

332

'operation after transaction started')

333

self.hookargs.update(hookargs)

333

self.hookargs.update(hookargs)

334

335

class TransactionUnavailable(RuntimeError):

335

class TransactionUnavailable(RuntimeError):

336

pass

336

pass

337

338

def _notransaction():

338

def _notransaction():

339

"""default method to get a transaction while processing a bundle

339

"""default method to get a transaction while processing a bundle

340

341

Raise an exception to highlight the fact that no transaction was expected

341

Raise an exception to highlight the fact that no transaction was expected

342

to be created"""

342

to be created"""

343

raise TransactionUnavailable()

343

raise TransactionUnavailable()

344

345

def applybundle(repo, unbundler, tr, source, url=None, **kwargs):

345

def applybundle(repo, unbundler, tr, source, url=None, **kwargs):

346

# transform me into unbundler.apply() as soon as the freeze is lifted

346

# transform me into unbundler.apply() as soon as the freeze is lifted

347

if isinstance(unbundler, unbundle20):

347

if isinstance(unbundler, unbundle20):

348

tr.hookargs['bundle2'] = '1'

348

tr.hookargs['bundle2'] = '1'

349

if source is not None and 'source' not in tr.hookargs:

349

if source is not None and 'source' not in tr.hookargs:

350

tr.hookargs['source'] = source

350

tr.hookargs['source'] = source

351

if url is not None and 'url' not in tr.hookargs:

351

if url is not None and 'url' not in tr.hookargs:

352

tr.hookargs['url'] = url

352

tr.hookargs['url'] = url

353

return processbundle(repo, unbundler, lambda: tr, source=source)

353

return processbundle(repo, unbundler, lambda: tr, source=source)

354

else:

354

else:

355

# the transactiongetter won't be used, but we might as well set it

355

# the transactiongetter won't be used, but we might as well set it

356

op = bundleoperation(repo, lambda: tr, source=source)

356

op = bundleoperation(repo, lambda: tr, source=source)

357

_processchangegroup(op, unbundler, tr, source, url, **kwargs)

357

_processchangegroup(op, unbundler, tr, source, url, **kwargs)

358

return op

358

return op

359

360

class partiterator(object):

360

class partiterator(object):

361

def __init__(self, repo, op, unbundler):

361

def __init__(self, repo, op, unbundler):

362

self.repo = repo

362

self.repo = repo

363

self.op = op

363

self.op = op

364

self.unbundler = unbundler

364

self.unbundler = unbundler

365

self.iterator = None

365

self.iterator = None

366

self.count = 0

366

self.count = 0

367

self.current = None

367

self.current = None

368

369

def __enter__(self):

369

def __enter__(self):

370

def func():

370

def func():

371

itr = enumerate(self.unbundler.iterparts())

371

itr = enumerate(self.unbundler.iterparts())

372

for count, p in itr:

372

for count, p in itr:

373

self.count = count

373

self.count = count

374

self.current = p

374

self.current = p

375

yield p

375

yield p

376

p.consume()

376

p.consume()

377

self.current = None

377

self.current = None

378

self.iterator = func()

378

self.iterator = func()

379

return self.iterator

379

return self.iterator

380

381

def __exit__(self, type, exc, tb):

381

def __exit__(self, type, exc, tb):

382

if not self.iterator:

382

if not self.iterator:

383

return

383

return

384

385

# Only gracefully abort in a normal exception situation. User aborts

385

# Only gracefully abort in a normal exception situation. User aborts

386

# like Ctrl+C throw a KeyboardInterrupt which is not a base Exception,

386

# like Ctrl+C throw a KeyboardInterrupt which is not a base Exception,

387

# and should not gracefully cleanup.

387

# and should not gracefully cleanup.

388

if isinstance(exc, Exception):

388

if isinstance(exc, Exception):

389

# Any exceptions seeking to the end of the bundle at this point are

389

# Any exceptions seeking to the end of the bundle at this point are

390

# almost certainly related to the underlying stream being bad.

390

# almost certainly related to the underlying stream being bad.

391

# And, chances are that the exception we're handling is related to

391

# And, chances are that the exception we're handling is related to

392

# getting in that bad state. So, we swallow the seeking error and

392

# getting in that bad state. So, we swallow the seeking error and

393

# re-raise the original error.

393

# re-raise the original error.

394

seekerror = False

394

seekerror = False

395

try:

395

try:

396

if self.current:

396

if self.current:

397

# consume the part content to not corrupt the stream.

397

# consume the part content to not corrupt the stream.

398

self.current.consume()

398

self.current.consume()

399

400

for part in self.iterator:

400

for part in self.iterator:

401

# consume the bundle content

401

# consume the bundle content

402

part.consume()

402

part.consume()

403

except Exception:

403

except Exception:

404

seekerror = True

404

seekerror = True

405

406

# Small hack to let caller code distinguish exceptions from bundle2

406

# Small hack to let caller code distinguish exceptions from bundle2

407

# processing from processing the old format. This is mostly needed

407

# processing from processing the old format. This is mostly needed

408

# to handle different return codes to unbundle according to the type

408

# to handle different return codes to unbundle according to the type

409

# of bundle. We should probably clean up or drop this return code

409

# of bundle. We should probably clean up or drop this return code

410

# craziness in a future version.

410

# craziness in a future version.

411

exc.duringunbundle2 = True

411

exc.duringunbundle2 = True

412

salvaged = []

412

salvaged = []

413

replycaps = None

413

replycaps = None

414

if self.op.reply is not None:

414

if self.op.reply is not None:

415

salvaged = self.op.reply.salvageoutput()

415

salvaged = self.op.reply.salvageoutput()

416

replycaps = self.op.reply.capabilities

416

replycaps = self.op.reply.capabilities

417

exc._replycaps = replycaps

417

exc._replycaps = replycaps

418

exc._bundle2salvagedoutput = salvaged

418

exc._bundle2salvagedoutput = salvaged

419

420

# Re-raising from a variable loses the original stack. So only use

420

# Re-raising from a variable loses the original stack. So only use

421

# that form if we need to.

421

# that form if we need to.

422

if seekerror:

422

if seekerror:

423

raise exc

423

raise exc

424

425

self.repo.ui.debug('bundle2-input-bundle: %i parts total\n' %

425

self.repo.ui.debug('bundle2-input-bundle: %i parts total\n' %

426

self.count)

426

self.count)

427

428

def processbundle(repo, unbundler, transactiongetter=None, op=None, source=''):

428

def processbundle(repo, unbundler, transactiongetter=None, op=None, source=''):

429

"""This function process a bundle, apply effect to/from a repo

429

"""This function process a bundle, apply effect to/from a repo

430

431

It iterates over each part then searches for and uses the proper handling

431

It iterates over each part then searches for and uses the proper handling

432

code to process the part. Parts are processed in order.

432

code to process the part. Parts are processed in order.

433

434

Unknown Mandatory part will abort the process.

434

Unknown Mandatory part will abort the process.

435

436

It is temporarily possible to provide a prebuilt bundleoperation to the

436

It is temporarily possible to provide a prebuilt bundleoperation to the

437

function. This is used to ensure output is properly propagated in case of

437

function. This is used to ensure output is properly propagated in case of

438

an error during the unbundling. This output capturing part will likely be

438

an error during the unbundling. This output capturing part will likely be

439

reworked and this ability will probably go away in the process.

439

reworked and this ability will probably go away in the process.

440

"""

440

"""

441

if op is None:

441

if op is None:

442

if transactiongetter is None:

442

if transactiongetter is None:

443

transactiongetter = _notransaction

443

transactiongetter = _notransaction

444

op = bundleoperation(repo, transactiongetter, source=source)

444

op = bundleoperation(repo, transactiongetter, source=source)

445

# todo:

445

# todo:

446

# - replace this is a init function soon.

446

# - replace this is a init function soon.

447

# - exception catching

447

# - exception catching

448

unbundler.params

448

unbundler.params

449

if repo.ui.debugflag:

449

if repo.ui.debugflag:

450

msg = ['bundle2-input-bundle:']

450

msg = ['bundle2-input-bundle:']

451

if unbundler.params:

451

if unbundler.params:

452

msg.append(' %i params' % len(unbundler.params))

452

msg.append(' %i params' % len(unbundler.params))

453

if op._gettransaction is None or op._gettransaction is _notransaction:

453

if op._gettransaction is None or op._gettransaction is _notransaction:

454

msg.append(' no-transaction')

454

msg.append(' no-transaction')

455

else:

455

else:

456

msg.append(' with-transaction')

456

msg.append(' with-transaction')

457

msg.append('\n')

457

msg.append('\n')

458

repo.ui.debug(''.join(msg))

458

repo.ui.debug(''.join(msg))

459

460

processparts(repo, op, unbundler)

460

processparts(repo, op, unbundler)

461

462

return op

462

return op

463

464

def processparts(repo, op, unbundler):

464

def processparts(repo, op, unbundler):

465

with partiterator(repo, op, unbundler) as parts:

465

with partiterator(repo, op, unbundler) as parts:

466

for part in parts:

466

for part in parts:

467

_processpart(op, part)

467

_processpart(op, part)

468

469

def _processchangegroup(op, cg, tr, source, url, **kwargs):

469

def _processchangegroup(op, cg, tr, source, url, **kwargs):

470

ret = cg.apply(op.repo, tr, source, url, **kwargs)

470

ret = cg.apply(op.repo, tr, source, url, **kwargs)

471

op.records.add('changegroup', {

471

op.records.add('changegroup', {

472

'return': ret,

472

'return': ret,

473

})

473

})

474

return ret

474

return ret

475

476

def _gethandler(op, part):

476

def _gethandler(op, part):

477

status = 'unknown' # used by debug output

477

status = 'unknown' # used by debug output

478

try:

478

try:

479

handler = parthandlermapping.get(part.type)

479

handler = parthandlermapping.get(part.type)

480

if handler is None:

480

if handler is None:

481

status = 'unsupported-type'

481

status = 'unsupported-type'

482

raise error.BundleUnknownFeatureError(parttype=part.type)

482

raise error.BundleUnknownFeatureError(parttype=part.type)

483

indebug(op.ui, 'found a handler for part %s' % part.type)

483

indebug(op.ui, 'found a handler for part %s' % part.type)

484

unknownparams = part.mandatorykeys - handler.params

484

unknownparams = part.mandatorykeys - handler.params

485

if unknownparams:

485

if unknownparams:

486

unknownparams = list(unknownparams)

486

unknownparams = list(unknownparams)

487

unknownparams.sort()

487

unknownparams.sort()

488

status = 'unsupported-params (%s)' % ', '.join(unknownparams)

488

status = 'unsupported-params (%s)' % ', '.join(unknownparams)

489

raise error.BundleUnknownFeatureError(parttype=part.type,

489

raise error.BundleUnknownFeatureError(parttype=part.type,

490

params=unknownparams)

490

params=unknownparams)

491

status = 'supported'

491

status = 'supported'

492

except error.BundleUnknownFeatureError as exc:

492

except error.BundleUnknownFeatureError as exc:

493

if part.mandatory: # mandatory parts

493

if part.mandatory: # mandatory parts

494

raise

494

raise

495

indebug(op.ui, 'ignoring unsupported advisory part %s' % exc)

495

indebug(op.ui, 'ignoring unsupported advisory part %s' % exc)

496

return # skip to part processing

496

return # skip to part processing

497

finally:

497

finally:

498

if op.ui.debugflag:

498

if op.ui.debugflag:

499

msg = ['bundle2-input-part: "%s"' % part.type]

499

msg = ['bundle2-input-part: "%s"' % part.type]

500

if not part.mandatory:

500

if not part.mandatory:

501

msg.append(' (advisory)')

501

msg.append(' (advisory)')

502

nbmp = len(part.mandatorykeys)

502

nbmp = len(part.mandatorykeys)

503

nbap = len(part.params) - nbmp

503

nbap = len(part.params) - nbmp

504

if nbmp or nbap:

504

if nbmp or nbap:

505

msg.append(' (params:')

505

msg.append(' (params:')

506

if nbmp:

506

if nbmp:

507

msg.append(' %i mandatory' % nbmp)

507

msg.append(' %i mandatory' % nbmp)

508

if nbap:

508

if nbap:

509

msg.append(' %i advisory' % nbmp)

509

msg.append(' %i advisory' % nbmp)

510

msg.append(')')

510

msg.append(')')

511

msg.append(' %s\n' % status)

511

msg.append(' %s\n' % status)

512

op.ui.debug(''.join(msg))

512

op.ui.debug(''.join(msg))

513

514

return handler

514

return handler

515

516

def _processpart(op, part):

516

def _processpart(op, part):

517

"""process a single part from a bundle

517

"""process a single part from a bundle

518

519

The part is guaranteed to have been fully consumed when the function exits

519

The part is guaranteed to have been fully consumed when the function exits

520

(even if an exception is raised)."""

520

(even if an exception is raised)."""

521

handler = _gethandler(op, part)

521

handler = _gethandler(op, part)

522

if handler is None:

522

if handler is None:

523

return

523

return

524

525

# handler is called outside the above try block so that we don't

525

# handler is called outside the above try block so that we don't

526

# risk catching KeyErrors from anything other than the

526

# risk catching KeyErrors from anything other than the

527

# parthandlermapping lookup (any KeyError raised by handler()

527

# parthandlermapping lookup (any KeyError raised by handler()

528

# itself represents a defect of a different variety).

528

# itself represents a defect of a different variety).

529

output = None

529

output = None

530

if op.captureoutput and op.reply is not None:

530

if op.captureoutput and op.reply is not None:

531

op.ui.pushbuffer(error=True, subproc=True)

531

op.ui.pushbuffer(error=True, subproc=True)

532

output = ''

532

output = ''

533

try:

533

try:

534

handler(op, part)

534

handler(op, part)

535

finally:

535

finally:

536

if output is not None:

536

if output is not None:

537

output = op.ui.popbuffer()

537

output = op.ui.popbuffer()

538

if output:

538

if output:

539

outpart = op.reply.newpart('output', data=output,

539

outpart = op.reply.newpart('output', data=output,

540

mandatory=False)

540

mandatory=False)

541

outpart.addparam(

541

outpart.addparam(

542

'in-reply-to', pycompat.bytestr(part.id), mandatory=False)

542

'in-reply-to', pycompat.bytestr(part.id), mandatory=False)

543

544

def decodecaps(blob):

544

def decodecaps(blob):

545

"""decode a bundle2 caps bytes blob into a dictionary

545

"""decode a bundle2 caps bytes blob into a dictionary

546

547

The blob is a list of capabilities (one per line)

547

The blob is a list of capabilities (one per line)

548

Capabilities may have values using a line of the form::

548

Capabilities may have values using a line of the form::

549

550

capability=value1,value2,value3

550

capability=value1,value2,value3

551

552

The values are always a list."""

552

The values are always a list."""

553

caps = {}

553

caps = {}

554

for line in blob.splitlines():

554

for line in blob.splitlines():

555

if not line:

555

if not line:

556

continue

556

continue

557

if '=' not in line:

557

if '=' not in line:

558

key, vals = line, ()

558

key, vals = line, ()

559

else:

559

else:

560

key, vals = line.split('=', 1)

560

key, vals = line.split('=', 1)

561

vals = vals.split(',')

561

vals = vals.split(',')

562

key = urlreq.unquote(key)

562

key = urlreq.unquote(key)

563

vals = [urlreq.unquote(v) for v in vals]

563

vals = [urlreq.unquote(v) for v in vals]

564

caps[key] = vals

564

caps[key] = vals

565

return caps

565

return caps

566

567

def encodecaps(caps):

567

def encodecaps(caps):

568

"""encode a bundle2 caps dictionary into a bytes blob"""

568

"""encode a bundle2 caps dictionary into a bytes blob"""

569

chunks = []

569

chunks = []

570

for ca in sorted(caps):

570

for ca in sorted(caps):

571

vals = caps[ca]

571

vals = caps[ca]

572

ca = urlreq.quote(ca)

572

ca = urlreq.quote(ca)

573

vals = [urlreq.quote(v) for v in vals]

573

vals = [urlreq.quote(v) for v in vals]

574

if vals:

574

if vals:

575

ca = "%s=%s" % (ca, ','.join(vals))

575

ca = "%s=%s" % (ca, ','.join(vals))

576

chunks.append(ca)

576

chunks.append(ca)

577

return '\n'.join(chunks)

577

return '\n'.join(chunks)

578

579

bundletypes = {

579

bundletypes = {

580

"": ("", 'UN'), # only when using unbundle on ssh and old http servers

580

"": ("", 'UN'), # only when using unbundle on ssh and old http servers

581

# since the unification ssh accepts a header but there

581

# since the unification ssh accepts a header but there

582

# is no capability signaling it.

582

# is no capability signaling it.

583

"HG20": (), # special-cased below

583

"HG20": (), # special-cased below

584

"HG10UN": ("HG10UN", 'UN'),

584

"HG10UN": ("HG10UN", 'UN'),

585

"HG10BZ": ("HG10", 'BZ'),

585

"HG10BZ": ("HG10", 'BZ'),

586

"HG10GZ": ("HG10GZ", 'GZ'),

586

"HG10GZ": ("HG10GZ", 'GZ'),

587

}

587

}

588

589

# hgweb uses this list to communicate its preferred type

589

# hgweb uses this list to communicate its preferred type

590

bundlepriority = ['HG10GZ', 'HG10BZ', 'HG10UN']

590

bundlepriority = ['HG10GZ', 'HG10BZ', 'HG10UN']

591

592

class bundle20(object):

592

class bundle20(object):

593

"""represent an outgoing bundle2 container

593

"""represent an outgoing bundle2 container

594

595

Use the `addparam` method to add stream level parameter. and `newpart` to

595

Use the `addparam` method to add stream level parameter. and `newpart` to

596

populate it. Then call `getchunks` to retrieve all the binary chunks of

596

populate it. Then call `getchunks` to retrieve all the binary chunks of

597

data that compose the bundle2 container."""

597

data that compose the bundle2 container."""

598

599

_magicstring = 'HG20'

599

_magicstring = 'HG20'

600

601

def __init__(self, ui, capabilities=()):

601

def __init__(self, ui, capabilities=()):

602

self.ui = ui

602

self.ui = ui

603

self._params = []

603

self._params = []

604

self._parts = []

604

self._parts = []

605

self.capabilities = dict(capabilities)

605

self.capabilities = dict(capabilities)

606

self._compengine = util.compengines.forbundletype('UN')

606

self._compengine = util.compengines.forbundletype('UN')

607

self._compopts = None

607

self._compopts = None

608

# If compression is being handled by a consumer of the raw

608

# If compression is being handled by a consumer of the raw

609

# data (e.g. the wire protocol), unsetting this flag tells

609

# data (e.g. the wire protocol), unsetting this flag tells

610

# consumers that the bundle is best left uncompressed.

610

# consumers that the bundle is best left uncompressed.

611

self.prefercompressed = True

611

self.prefercompressed = True

612

613

def setcompression(self, alg, compopts=None):

613

def setcompression(self, alg, compopts=None):

614

"""setup core part compression to <alg>"""

614

"""setup core part compression to <alg>"""

615

if alg in (None, 'UN'):

615

if alg in (None, 'UN'):

616

return

616

return

617

assert not any(n.lower() == 'compression' for n, v in self._params)

617

assert not any(n.lower() == 'compression' for n, v in self._params)

618

self.addparam('Compression', alg)

618

self.addparam('Compression', alg)

619

self._compengine = util.compengines.forbundletype(alg)

619

self._compengine = util.compengines.forbundletype(alg)

620

self._compopts = compopts

620

self._compopts = compopts

621

622

@property

622

@property

623

def nbparts(self):

623

def nbparts(self):

624

"""total number of parts added to the bundler"""

624

"""total number of parts added to the bundler"""

625

return len(self._parts)

625

return len(self._parts)

626

627

# methods used to defines the bundle2 content

627

# methods used to defines the bundle2 content

628

def addparam(self, name, value=None):

628

def addparam(self, name, value=None):

629

"""add a stream level parameter"""

629

"""add a stream level parameter"""

630

if not name:

630

if not name:

631

raise error.ProgrammingError(b'empty parameter name')

631

raise error.ProgrammingError(b'empty parameter name')

632

if name[0:1] not in pycompat.bytestr(string.ascii_letters):

632

if name[0:1] not in pycompat.bytestr(string.ascii_letters):

633

raise error.ProgrammingError(b'non letter first character: %s'

633

raise error.ProgrammingError(b'non letter first character: %s'

634

% name)

634

% name)

635

self._params.append((name, value))

635

self._params.append((name, value))

636

637

def addpart(self, part):

637

def addpart(self, part):

638

"""add a new part to the bundle2 container

638

"""add a new part to the bundle2 container

639

640

Parts contains the actual applicative payload."""

640

Parts contains the actual applicative payload."""

641

assert part.id is None

641

assert part.id is None

642

part.id = len(self._parts) # very cheap counter

642

part.id = len(self._parts) # very cheap counter

643

self._parts.append(part)

643

self._parts.append(part)

644

645

def newpart(self, typeid, *args, **kwargs):

645

def newpart(self, typeid, *args, **kwargs):

646

"""create a new part and add it to the containers

646

"""create a new part and add it to the containers

647

648

As the part is directly added to the containers. For now, this means

648

As the part is directly added to the containers. For now, this means

649

that any failure to properly initialize the part after calling

649

that any failure to properly initialize the part after calling

650

``newpart`` should result in a failure of the whole bundling process.

650

``newpart`` should result in a failure of the whole bundling process.

651

652

You can still fall back to manually create and add if you need better

652

You can still fall back to manually create and add if you need better

653

control."""

653

control."""

654

part = bundlepart(typeid, *args, **kwargs)

654

part = bundlepart(typeid, *args, **kwargs)

655

self.addpart(part)

655

self.addpart(part)

656

return part

656

return part

657

658

# methods used to generate the bundle2 stream

658

# methods used to generate the bundle2 stream

659

def getchunks(self):

659

def getchunks(self):

660

if self.ui.debugflag:

660

if self.ui.debugflag:

661

msg = ['bundle2-output-bundle: "%s",' % self._magicstring]

661

msg = ['bundle2-output-bundle: "%s",' % self._magicstring]

662

if self._params:

662

if self._params:

663

msg.append(' (%i params)' % len(self._params))

663

msg.append(' (%i params)' % len(self._params))

664

msg.append(' %i parts total\n' % len(self._parts))

664

msg.append(' %i parts total\n' % len(self._parts))

665

self.ui.debug(''.join(msg))

665

self.ui.debug(''.join(msg))

666

outdebug(self.ui, 'start emission of %s stream' % self._magicstring)

666

outdebug(self.ui, 'start emission of %s stream' % self._magicstring)

667

yield self._magicstring

667

yield self._magicstring

668

param = self._paramchunk()

668

param = self._paramchunk()

669

outdebug(self.ui, 'bundle parameter: %s' % param)

669

outdebug(self.ui, 'bundle parameter: %s' % param)

670

yield _pack(_fstreamparamsize, len(param))

670

yield _pack(_fstreamparamsize, len(param))

671

if param:

671

if param:

672

yield param

672

yield param

673

for chunk in self._compengine.compressstream(self._getcorechunk(),

673

for chunk in self._compengine.compressstream(self._getcorechunk(),

674

self._compopts):

674

self._compopts):

675

yield chunk

675

yield chunk

676

677

def _paramchunk(self):

677

def _paramchunk(self):

678

"""return a encoded version of all stream parameters"""

678

"""return a encoded version of all stream parameters"""

679

blocks = []

679

blocks = []

680

for par, value in self._params:

680

for par, value in self._params:

681

par = urlreq.quote(par)

681

par = urlreq.quote(par)

682

if value is not None:

682

if value is not None:

683

value = urlreq.quote(value)

683

value = urlreq.quote(value)

684

par = '%s=%s' % (par, value)

684

par = '%s=%s' % (par, value)

685

blocks.append(par)

685

blocks.append(par)

686

return ' '.join(blocks)

686

return ' '.join(blocks)

687

688

def _getcorechunk(self):

688

def _getcorechunk(self):

689

"""yield chunk for the core part of the bundle

689

"""yield chunk for the core part of the bundle

690

691

(all but headers and parameters)"""

691

(all but headers and parameters)"""

692

outdebug(self.ui, 'start of parts')

692

outdebug(self.ui, 'start of parts')

693

for part in self._parts:

693

for part in self._parts:

694

outdebug(self.ui, 'bundle part: "%s"' % part.type)

694

outdebug(self.ui, 'bundle part: "%s"' % part.type)

695

for chunk in part.getchunks(ui=self.ui):

695

for chunk in part.getchunks(ui=self.ui):

696

yield chunk

696

yield chunk

697

outdebug(self.ui, 'end of bundle')

697

outdebug(self.ui, 'end of bundle')

698

yield _pack(_fpartheadersize, 0)

698

yield _pack(_fpartheadersize, 0)

699

700

701

def salvageoutput(self):

701

def salvageoutput(self):

702

"""return a list with a copy of all output parts in the bundle

702

"""return a list with a copy of all output parts in the bundle

703

704

This is meant to be used during error handling to make sure we preserve

704

This is meant to be used during error handling to make sure we preserve

705

server output"""

705

server output"""

706

salvaged = []

706

salvaged = []

707

for part in self._parts:

707

for part in self._parts:

708

if part.type.startswith('output'):

708

if part.type.startswith('output'):

709

salvaged.append(part.copy())

709

salvaged.append(part.copy())

710

return salvaged

710

return salvaged

711

712

713

class unpackermixin(object):

713

class unpackermixin(object):

714

"""A mixin to extract bytes and struct data from a stream"""

714

"""A mixin to extract bytes and struct data from a stream"""

715

716

def __init__(self, fp):

716

def __init__(self, fp):

717

self._fp = fp

717

self._fp = fp

718

719

def _unpack(self, format):

719

def _unpack(self, format):

720

"""unpack this struct format from the stream

720

"""unpack this struct format from the stream

721

722

This method is meant for internal usage by the bundle2 protocol only.

722

This method is meant for internal usage by the bundle2 protocol only.

723

They directly manipulate the low level stream including bundle2 level

723

They directly manipulate the low level stream including bundle2 level

724

instruction.

724

instruction.

725

726

Do not use it to implement higher-level logic or methods."""

726

Do not use it to implement higher-level logic or methods."""

727

data = self._readexact(struct.calcsize(format))

727

data = self._readexact(struct.calcsize(format))

728

return _unpack(format, data)

728

return _unpack(format, data)

729

730

def _readexact(self, size):

730

def _readexact(self, size):

731

"""read exactly <size> bytes from the stream

731

"""read exactly <size> bytes from the stream

732

733

This method is meant for internal usage by the bundle2 protocol only.

733

This method is meant for internal usage by the bundle2 protocol only.

734

They directly manipulate the low level stream including bundle2 level

734

They directly manipulate the low level stream including bundle2 level

735

instruction.

735

instruction.

736

737

Do not use it to implement higher-level logic or methods."""

737

Do not use it to implement higher-level logic or methods."""

738

return changegroup.readexactly(self._fp, size)

738

return changegroup.readexactly(self._fp, size)

739

740

def getunbundler(ui, fp, magicstring=None):

740

def getunbundler(ui, fp, magicstring=None):

741

"""return a valid unbundler object for a given magicstring"""

741

"""return a valid unbundler object for a given magicstring"""

742

if magicstring is None:

742

if magicstring is None:

743

magicstring = changegroup.readexactly(fp, 4)

743

magicstring = changegroup.readexactly(fp, 4)

744

magic, version = magicstring[0:2], magicstring[2:4]

744

magic, version = magicstring[0:2], magicstring[2:4]

745

if magic != 'HG':

745

if magic != 'HG':

746

ui.debug(

746

ui.debug(

747

"error: invalid magic: %r (version %r), should be 'HG'\n"

747

"error: invalid magic: %r (version %r), should be 'HG'\n"

748

% (magic, version))

748

% (magic, version))

749

raise error.Abort(_('not a Mercurial bundle'))

749

raise error.Abort(_('not a Mercurial bundle'))

750

unbundlerclass = formatmap.get(version)

750

unbundlerclass = formatmap.get(version)

751

if unbundlerclass is None:

751

if unbundlerclass is None:

752

raise error.Abort(_('unknown bundle version %s') % version)

752

raise error.Abort(_('unknown bundle version %s') % version)

753

unbundler = unbundlerclass(ui, fp)

753

unbundler = unbundlerclass(ui, fp)

754

indebug(ui, 'start processing of %s stream' % magicstring)

754

indebug(ui, 'start processing of %s stream' % magicstring)

755

return unbundler

755

return unbundler

756

757

class unbundle20(unpackermixin):

757

class unbundle20(unpackermixin):

758

"""interpret a bundle2 stream

758

"""interpret a bundle2 stream

759

760

This class is fed with a binary stream and yields parts through its

760

This class is fed with a binary stream and yields parts through its

761

`iterparts` methods."""

761

`iterparts` methods."""

762

763

_magicstring = 'HG20'

763

_magicstring = 'HG20'

764

765

def __init__(self, ui, fp):

765

def __init__(self, ui, fp):

766

"""If header is specified, we do not read it out of the stream."""

766

"""If header is specified, we do not read it out of the stream."""

767

self.ui = ui

767

self.ui = ui

768

self._compengine = util.compengines.forbundletype('UN')

768

self._compengine = util.compengines.forbundletype('UN')

769

self._compressed = None

769

self._compressed = None

770

super(unbundle20, self).__init__(fp)

770

super(unbundle20, self).__init__(fp)

771

772

@util.propertycache

772

@util.propertycache

773

def params(self):

773

def params(self):

774

"""dictionary of stream level parameters"""

774

"""dictionary of stream level parameters"""

775

indebug(self.ui, 'reading bundle2 stream parameters')

775

indebug(self.ui, 'reading bundle2 stream parameters')

776

params = {}

776

params = {}

777

paramssize = self._unpack(_fstreamparamsize)[0]

777

paramssize = self._unpack(_fstreamparamsize)[0]

778

if paramssize < 0:

778

if paramssize < 0:

779

raise error.BundleValueError('negative bundle param size: %i'

779

raise error.BundleValueError('negative bundle param size: %i'

780

% paramssize)

780

% paramssize)

781

if paramssize:

781

if paramssize:

782

params = self._readexact(paramssize)

782

params = self._readexact(paramssize)

783

params = self._processallparams(params)

783

params = self._processallparams(params)

784

return params

784

return params

785

786

def _processallparams(self, paramsblock):

786

def _processallparams(self, paramsblock):

787

""""""

787

""""""

788

params = util.sortdict()

788

params = util.sortdict()

789

for p in paramsblock.split(' '):

789

for p in paramsblock.split(' '):

790

p = p.split('=', 1)

790

p = p.split('=', 1)

791

p = [urlreq.unquote(i) for i in p]

791

p = [urlreq.unquote(i) for i in p]

792

if len(p) < 2:

792

if len(p) < 2:

793

p.append(None)

793

p.append(None)

794

self._processparam(*p)

794

self._processparam(*p)

795

params[p[0]] = p[1]

795

params[p[0]] = p[1]

796

return params

796

return params

797

798

799

def _processparam(self, name, value):

799

def _processparam(self, name, value):

800

"""process a parameter, applying its effect if needed

800

"""process a parameter, applying its effect if needed

801

802

Parameter starting with a lower case letter are advisory and will be

802

Parameter starting with a lower case letter are advisory and will be

803

ignored when unknown. Those starting with an upper case letter are

803

ignored when unknown. Those starting with an upper case letter are

804

mandatory and will this function will raise a KeyError when unknown.

804

mandatory and will this function will raise a KeyError when unknown.

805

806

Note: no option are currently supported. Any input will be either

806

Note: no option are currently supported. Any input will be either

807

ignored or failing.

807

ignored or failing.

808

"""

808

"""

809

if not name:

809

if not name:

810

raise ValueError(r'empty parameter name')

810

raise ValueError(r'empty parameter name')

811

if name[0:1] not in pycompat.bytestr(string.ascii_letters):

811

if name[0:1] not in pycompat.bytestr(string.ascii_letters):

812

raise ValueError(r'non letter first character: %s' % name)

812

raise ValueError(r'non letter first character: %s' % name)

813

try:

813

try:

814

handler = b2streamparamsmap[name.lower()]

814

handler = b2streamparamsmap[name.lower()]

815

except KeyError:

815

except KeyError:

816

if name[0:1].islower():

816

if name[0:1].islower():

817

indebug(self.ui, "ignoring unknown parameter %s" % name)

817

indebug(self.ui, "ignoring unknown parameter %s" % name)

818

else:

818

else:

819

raise error.BundleUnknownFeatureError(params=(name,))

819

raise error.BundleUnknownFeatureError(params=(name,))

820

else:

820

else:

821

handler(self, name, value)

821

handler(self, name, value)

822

823

def _forwardchunks(self):

823

def _forwardchunks(self):

824

"""utility to transfer a bundle2 as binary

824

"""utility to transfer a bundle2 as binary

825

826

This is made necessary by the fact the 'getbundle' command over 'ssh'

826

This is made necessary by the fact the 'getbundle' command over 'ssh'

827

have no way to know then the reply end, relying on the bundle to be

827

have no way to know then the reply end, relying on the bundle to be

828

interpreted to know its end. This is terrible and we are sorry, but we

828

interpreted to know its end. This is terrible and we are sorry, but we

829

needed to move forward to get general delta enabled.

829

needed to move forward to get general delta enabled.

830

"""

830

"""

831

yield self._magicstring

831

yield self._magicstring

832

assert 'params' not in vars(self)

832

assert 'params' not in vars(self)

833

paramssize = self._unpack(_fstreamparamsize)[0]

833

paramssize = self._unpack(_fstreamparamsize)[0]

834

if paramssize < 0:

834

if paramssize < 0:

835

raise error.BundleValueError('negative bundle param size: %i'

835

raise error.BundleValueError('negative bundle param size: %i'

836

% paramssize)

836

% paramssize)

837

yield _pack(_fstreamparamsize, paramssize)

837

yield _pack(_fstreamparamsize, paramssize)

838

if paramssize:

838

if paramssize:

839

params = self._readexact(paramssize)

839

params = self._readexact(paramssize)

840

self._processallparams(params)

840

self._processallparams(params)

841

yield params

841

yield params

842

assert self._compengine.bundletype == 'UN'

842

assert self._compengine.bundletype == 'UN'

843

# From there, payload might need to be decompressed

843

# From there, payload might need to be decompressed

844

self._fp = self._compengine.decompressorreader(self._fp)

844

self._fp = self._compengine.decompressorreader(self._fp)

845

emptycount = 0

845

emptycount = 0

846

while emptycount < 2:

846

while emptycount < 2:

847

# so we can brainlessly loop

847

# so we can brainlessly loop

848

assert _fpartheadersize == _fpayloadsize

848

assert _fpartheadersize == _fpayloadsize

849

size = self._unpack(_fpartheadersize)[0]

849

size = self._unpack(_fpartheadersize)[0]

850

yield _pack(_fpartheadersize, size)

850

yield _pack(_fpartheadersize, size)

851

if size:

851

if size:

852

emptycount = 0

852

emptycount = 0

853

else:

853

else:

854

emptycount += 1

854

emptycount += 1

855

continue

855

continue

856

if size == flaginterrupt:

856

if size == flaginterrupt:

857

continue

857

continue

858

elif size < 0:

858

elif size < 0:

859

raise error.BundleValueError('negative chunk size: %i')

859

raise error.BundleValueError('negative chunk size: %i')

860

yield self._readexact(size)

860

yield self._readexact(size)

861

862

863

def iterparts(self, seekable=False):

863

def iterparts(self, seekable=False):

864

"""yield all parts contained in the stream"""

864

"""yield all parts contained in the stream"""

865

cls = seekableunbundlepart if seekable else unbundlepart

865

cls = seekableunbundlepart if seekable else unbundlepart

866

# make sure param have been loaded

866

# make sure param have been loaded

867

self.params

867

self.params

868

# From there, payload need to be decompressed

868

# From there, payload need to be decompressed

869

self._fp = self._compengine.decompressorreader(self._fp)

869

self._fp = self._compengine.decompressorreader(self._fp)

870

indebug(self.ui, 'start extraction of bundle2 parts')

870

indebug(self.ui, 'start extraction of bundle2 parts')

871

headerblock = self._readpartheader()

871

headerblock = self._readpartheader()

872

while headerblock is not None:

872

while headerblock is not None:

873

part = cls(self.ui, headerblock, self._fp)

873

part = cls(self.ui, headerblock, self._fp)

874

yield part

874

yield part

875

# Ensure part is fully consumed so we can start reading the next

875

# Ensure part is fully consumed so we can start reading the next

876

# part.

876

# part.

877

part.consume()

877

part.consume()

878

879

headerblock = self._readpartheader()

879

headerblock = self._readpartheader()

880

indebug(self.ui, 'end of bundle2 stream')

880

indebug(self.ui, 'end of bundle2 stream')

881

882

def _readpartheader(self):

882

def _readpartheader(self):

883

"""reads a part header size and return the bytes blob

883

"""reads a part header size and return the bytes blob

884

885

returns None if empty"""

885

returns None if empty"""

886

headersize = self._unpack(_fpartheadersize)[0]

886

headersize = self._unpack(_fpartheadersize)[0]

887

if headersize < 0:

887

if headersize < 0:

888

raise error.BundleValueError('negative part header size: %i'

888

raise error.BundleValueError('negative part header size: %i'

889

% headersize)

889

% headersize)

890

indebug(self.ui, 'part header size: %i' % headersize)

890

indebug(self.ui, 'part header size: %i' % headersize)

891

if headersize:

891

if headersize:

892

return self._readexact(headersize)

892

return self._readexact(headersize)

893

return None

893

return None

894

895

def compressed(self):

895

def compressed(self):

896

self.params # load params

896

self.params # load params

897

return self._compressed

897

return self._compressed

898

899

def close(self):

899

def close(self):

900

"""close underlying file"""

900

"""close underlying file"""

901

if util.safehasattr(self._fp, 'close'):

901

if util.safehasattr(self._fp, 'close'):

902

return self._fp.close()

902

return self._fp.close()

903

904

formatmap = {'20': unbundle20}

904

formatmap = {'20': unbundle20}

905

906

b2streamparamsmap = {}

906

b2streamparamsmap = {}

907

908

def b2streamparamhandler(name):

908

def b2streamparamhandler(name):

909

"""register a handler for a stream level parameter"""

909

"""register a handler for a stream level parameter"""

910

def decorator(func):

910

def decorator(func):

911

assert name not in formatmap

911

assert name not in formatmap

912

b2streamparamsmap[name] = func

912

b2streamparamsmap[name] = func

913

return func

913

return func

914

return decorator

914

return decorator

915

916

@b2streamparamhandler('compression')

916

@b2streamparamhandler('compression')

917

def processcompression(unbundler, param, value):

917

def processcompression(unbundler, param, value):

918

"""read compression parameter and install payload decompression"""

918

"""read compression parameter and install payload decompression"""

919

if value not in util.compengines.supportedbundletypes:

919

if value not in util.compengines.supportedbundletypes:

920

raise error.BundleUnknownFeatureError(params=(param,),

920

raise error.BundleUnknownFeatureError(params=(param,),

921

values=(value,))

921

values=(value,))

922

unbundler._compengine = util.compengines.forbundletype(value)

922

unbundler._compengine = util.compengines.forbundletype(value)

923

if value is not None:

923

if value is not None:

924

unbundler._compressed = True

924

unbundler._compressed = True

925

926

class bundlepart(object):

926

class bundlepart(object):

927

"""A bundle2 part contains application level payload

927

"""A bundle2 part contains application level payload

928

929

The part `type` is used to route the part to the application level

929

The part `type` is used to route the part to the application level

930

handler.

930

handler.

931

932

The part payload is contained in ``part.data``. It could be raw bytes or a

932

The part payload is contained in ``part.data``. It could be raw bytes or a

933

generator of byte chunks.

933

generator of byte chunks.

934

935

You can add parameters to the part using the ``addparam`` method.

935

You can add parameters to the part using the ``addparam`` method.

936

Parameters can be either mandatory (default) or advisory. Remote side

936

Parameters can be either mandatory (default) or advisory. Remote side

937

should be able to safely ignore the advisory ones.

937

should be able to safely ignore the advisory ones.

938

939

Both data and parameters cannot be modified after the generation has begun.

939

Both data and parameters cannot be modified after the generation has begun.

940

"""

940

"""

941

942

def __init__(self, parttype, mandatoryparams=(), advisoryparams=(),

942

def __init__(self, parttype, mandatoryparams=(), advisoryparams=(),

943

data='', mandatory=True):

943

data='', mandatory=True):

944

validateparttype(parttype)

944

validateparttype(parttype)

945

self.id = None

945

self.id = None

946

self.type = parttype

946

self.type = parttype

947

self._data = data

947

self._data = data

948

self._mandatoryparams = list(mandatoryparams)

948

self._mandatoryparams = list(mandatoryparams)

949

self._advisoryparams = list(advisoryparams)

949

self._advisoryparams = list(advisoryparams)

950

# checking for duplicated entries

950

# checking for duplicated entries

951

self._seenparams = set()

951

self._seenparams = set()

952

for pname, __ in self._mandatoryparams + self._advisoryparams:

952

for pname, __ in self._mandatoryparams + self._advisoryparams:

953

if pname in self._seenparams:

953

if pname in self._seenparams:

954

raise error.ProgrammingError('duplicated params: %s' % pname)

954

raise error.ProgrammingError('duplicated params: %s' % pname)

955

self._seenparams.add(pname)

955

self._seenparams.add(pname)

956

# status of the part's generation:

956

# status of the part's generation:

957

# - None: not started,

957

# - None: not started,

958

# - False: currently generated,

958

# - False: currently generated,

959

# - True: generation done.

959

# - True: generation done.

960

self._generated = None

960

self._generated = None

961

self.mandatory = mandatory

961

self.mandatory = mandatory

962

963

def __repr__(self):

963

def __repr__(self):

964

cls = "%s.%s" % (self.__class__.__module__, self.__class__.__name__)

964

cls = "%s.%s" % (self.__class__.__module__, self.__class__.__name__)

965

return ('<%s object at %x; id: %s; type: %s; mandatory: %s>'

965

return ('<%s object at %x; id: %s; type: %s; mandatory: %s>'

966

% (cls, id(self), self.id, self.type, self.mandatory))

966

% (cls, id(self), self.id, self.type, self.mandatory))

967

968

def copy(self):

968

def copy(self):

969

"""return a copy of the part

969

"""return a copy of the part

970

971

The new part have the very same content but no partid assigned yet.

971

The new part have the very same content but no partid assigned yet.

972

Parts with generated data cannot be copied."""

972

Parts with generated data cannot be copied."""

973

assert not util.safehasattr(self.data, 'next')

973

assert not util.safehasattr(self.data, 'next')

974

return self.__class__(self.type, self._mandatoryparams,

974

return self.__class__(self.type, self._mandatoryparams,

975

self._advisoryparams, self._data, self.mandatory)

975

self._advisoryparams, self._data, self.mandatory)

976

977

# methods used to defines the part content

977

# methods used to defines the part content

978

@property

978

@property

979

def data(self):

979

def data(self):

980

return self._data

980

return self._data

981

982

@data.setter

982

@data.setter

983

def data(self, data):

983

def data(self, data):

984

if self._generated is not None:

984

if self._generated is not None:

985

raise error.ReadOnlyPartError('part is being generated')

985

raise error.ReadOnlyPartError('part is being generated')

986

self._data = data

986

self._data = data

987

988

@property

988

@property

989

def mandatoryparams(self):

989

def mandatoryparams(self):

990

# make it an immutable tuple to force people through ``addparam``

990

# make it an immutable tuple to force people through ``addparam``

991

return tuple(self._mandatoryparams)

991

return tuple(self._mandatoryparams)

992

993

@property

993

@property

994

def advisoryparams(self):

994

def advisoryparams(self):

995

# make it an immutable tuple to force people through ``addparam``

995

# make it an immutable tuple to force people through ``addparam``

996

return tuple(self._advisoryparams)

996

return tuple(self._advisoryparams)

997

998

def addparam(self, name, value='', mandatory=True):

998

def addparam(self, name, value='', mandatory=True):

999

"""add a parameter to the part

999

"""add a parameter to the part

1000

1001

If 'mandatory' is set to True, the remote handler must claim support

1001

If 'mandatory' is set to True, the remote handler must claim support

1002

for this parameter or the unbundling will be aborted.

1002

for this parameter or the unbundling will be aborted.

1003

1004

The 'name' and 'value' cannot exceed 255 bytes each.

1004

The 'name' and 'value' cannot exceed 255 bytes each.

1005

"""

1005

"""

1006

if self._generated is not None:

1006

if self._generated is not None:

1007

raise error.ReadOnlyPartError('part is being generated')

1007

raise error.ReadOnlyPartError('part is being generated')

1008

if name in self._seenparams:

1008

if name in self._seenparams:

1009

raise ValueError('duplicated params: %s' % name)

1009

raise ValueError('duplicated params: %s' % name)

1010

self._seenparams.add(name)

1010

self._seenparams.add(name)

1011

params = self._advisoryparams

1011

params = self._advisoryparams

1012

if mandatory:

1012

if mandatory:

1013

params = self._mandatoryparams

1013

params = self._mandatoryparams

1014

params.append((name, value))

1014

params.append((name, value))

1015

1016

# methods used to generates the bundle2 stream

1016

# methods used to generates the bundle2 stream

1017

def getchunks(self, ui):

1017

def getchunks(self, ui):

1018

if self._generated is not None:

1018

if self._generated is not None:

1019

raise error.ProgrammingError('part can only be consumed once')

1019

raise error.ProgrammingError('part can only be consumed once')

1020

self._generated = False

1020

self._generated = False

1021

1022

if ui.debugflag:

1022

if ui.debugflag:

1023

msg = ['bundle2-output-part: "%s"' % self.type]

1023

msg = ['bundle2-output-part: "%s"' % self.type]

1024

if not self.mandatory:

1024

if not self.mandatory:

1025

msg.append(' (advisory)')

1025

msg.append(' (advisory)')

1026

nbmp = len(self.mandatoryparams)

1026

nbmp = len(self.mandatoryparams)

1027

nbap = len(self.advisoryparams)

1027

nbap = len(self.advisoryparams)

1028

if nbmp or nbap:

1028

if nbmp or nbap:

1029

msg.append(' (params:')

1029

msg.append(' (params:')

1030

if nbmp:

1030

if nbmp:

1031

msg.append(' %i mandatory' % nbmp)

1031

msg.append(' %i mandatory' % nbmp)

1032

if nbap:

1032

if nbap:

1033

msg.append(' %i advisory' % nbmp)

1033

msg.append(' %i advisory' % nbmp)

1034

msg.append(')')

1034

msg.append(')')

1035

if not self.data:

1035

if not self.data:

1036

msg.append(' empty payload')

1036

msg.append(' empty payload')

1037

elif (util.safehasattr(self.data, 'next')

1037

elif (util.safehasattr(self.data, 'next')

1038

or util.safehasattr(self.data, '__next__')):

1038

or util.safehasattr(self.data, '__next__')):

1039

msg.append(' streamed payload')

1039

msg.append(' streamed payload')

1040

else:

1040

else:

1041

msg.append(' %i bytes payload' % len(self.data))

1041

msg.append(' %i bytes payload' % len(self.data))

1042

msg.append('\n')

1042

msg.append('\n')

1043

ui.debug(''.join(msg))

1043

ui.debug(''.join(msg))

1044

1045

#### header

1045

#### header

1046

if self.mandatory:

1046

if self.mandatory:

1047

parttype = self.type.upper()

1047

parttype = self.type.upper()

1048

else:

1048

else:

1049

parttype = self.type.lower()

1049

parttype = self.type.lower()

1050

outdebug(ui, 'part %s: "%s"' % (pycompat.bytestr(self.id), parttype))

1050

outdebug(ui, 'part %s: "%s"' % (pycompat.bytestr(self.id), parttype))

1051

## parttype

1051

## parttype

1052

header = [_pack(_fparttypesize, len(parttype)),

1052

header = [_pack(_fparttypesize, len(parttype)),

1053

parttype, _pack(_fpartid, self.id),

1053

parttype, _pack(_fpartid, self.id),

1054

]

1054

]

1055

## parameters

1055

## parameters

1056

# count

1056

# count

1057

manpar = self.mandatoryparams

1057

manpar = self.mandatoryparams

1058

advpar = self.advisoryparams

1058

advpar = self.advisoryparams

1059

header.append(_pack(_fpartparamcount, len(manpar), len(advpar)))

1059

header.append(_pack(_fpartparamcount, len(manpar), len(advpar)))

1060

# size

1060

# size

1061

parsizes = []

1061

parsizes = []

1062

for key, value in manpar:

1062

for key, value in manpar:

1063

parsizes.append(len(key))

1063

parsizes.append(len(key))

1064

parsizes.append(len(value))

1064

parsizes.append(len(value))

1065

for key, value in advpar:

1065

for key, value in advpar:

1066

parsizes.append(len(key))

1066

parsizes.append(len(key))

1067

parsizes.append(len(value))

1067

parsizes.append(len(value))

1068

paramsizes = _pack(_makefpartparamsizes(len(parsizes) // 2), *parsizes)

1068

paramsizes = _pack(_makefpartparamsizes(len(parsizes) // 2), *parsizes)

1069

header.append(paramsizes)

1069

header.append(paramsizes)

1070

# key, value

1070

# key, value

1071

for key, value in manpar:

1071

for key, value in manpar:

1072

header.append(key)

1072

header.append(key)

1073

header.append(value)

1073

header.append(value)

1074

for key, value in advpar:

1074

for key, value in advpar:

1075

header.append(key)

1075

header.append(key)

1076

header.append(value)

1076

header.append(value)

1077

## finalize header

1077

## finalize header

1078

try:

1078

try:

1079

headerchunk = ''.join(header)

1079

headerchunk = ''.join(header)

1080

except TypeError:

1080

except TypeError:

1081

raise TypeError(r'Found a non-bytes trying to '

1081

raise TypeError(r'Found a non-bytes trying to '

1082

r'build bundle part header: %r' % header)

1082

r'build bundle part header: %r' % header)

1083

outdebug(ui, 'header chunk size: %i' % len(headerchunk))

1083

outdebug(ui, 'header chunk size: %i' % len(headerchunk))

1084

yield _pack(_fpartheadersize, len(headerchunk))

1084

yield _pack(_fpartheadersize, len(headerchunk))

1085

yield headerchunk

1085

yield headerchunk

1086

## payload

1086

## payload

1087

try:

1087

try:

1088

for chunk in self._payloadchunks():

1088

for chunk in self._payloadchunks():

1089

outdebug(ui, 'payload chunk size: %i' % len(chunk))

1089

outdebug(ui, 'payload chunk size: %i' % len(chunk))

1090

yield _pack(_fpayloadsize, len(chunk))

1090

yield _pack(_fpayloadsize, len(chunk))

1091

yield chunk

1091

yield chunk

1092

except GeneratorExit:

1092

except GeneratorExit:

1093

# GeneratorExit means that nobody is listening for our

1093

# GeneratorExit means that nobody is listening for our

1094

# results anyway, so just bail quickly rather than trying

1094

# results anyway, so just bail quickly rather than trying

1095

# to produce an error part.

1095

# to produce an error part.

1096

ui.debug('bundle2-generatorexit\n')

1096

ui.debug('bundle2-generatorexit\n')

1097

raise

1097

raise

1098

except BaseException as exc:

1098

except BaseException as exc:

1099

bexc = stringutil.forcebytestr(exc)

1099

bexc = stringutil.forcebytestr(exc)

1100

# backup exception data for later

1100

# backup exception data for later

1101

ui.debug('bundle2-input-stream-interrupt: encoding exception %s'

1101

ui.debug('bundle2-input-stream-interrupt: encoding exception %s'

1102

% bexc)

1102

% bexc)

1103

tb = sys.exc_info()[2]

1103

tb = sys.exc_info()[2]

1104

msg = 'unexpected error: %s' % bexc

1104

msg = 'unexpected error: %s' % bexc

1105

interpart = bundlepart('error:abort', [('message', msg)],

1105

interpart = bundlepart('error:abort', [('message', msg)],

1106

mandatory=False)

1106

mandatory=False)

1107

interpart.id = 0

1107

interpart.id = 0

1108

yield _pack(_fpayloadsize, -1)

1108

yield _pack(_fpayloadsize, -1)

1109

for chunk in interpart.getchunks(ui=ui):

1109

for chunk in interpart.getchunks(ui=ui):

1110

yield chunk

1110

yield chunk

1111

outdebug(ui, 'closing payload chunk')

1111

outdebug(ui, 'closing payload chunk')

1112

# abort current part payload

1112

# abort current part payload

1113

yield _pack(_fpayloadsize, 0)

1113

yield _pack(_fpayloadsize, 0)

1114

pycompat.raisewithtb(exc, tb)

1114

pycompat.raisewithtb(exc, tb)

1115

# end of payload

1115

# end of payload

1116

outdebug(ui, 'closing payload chunk')

1116

outdebug(ui, 'closing payload chunk')

1117

yield _pack(_fpayloadsize, 0)

1117

yield _pack(_fpayloadsize, 0)

1118

self._generated = True

1118

self._generated = True

1119

1120

def _payloadchunks(self):

1120

def _payloadchunks(self):

1121

"""yield chunks of a the part payload

1121

"""yield chunks of a the part payload

1122

1123

Exists to handle the different methods to provide data to a part."""

1123

Exists to handle the different methods to provide data to a part."""

1124

# we only support fixed size data now.

1124

# we only support fixed size data now.

1125

# This will be improved in the future.

1125

# This will be improved in the future.

1126

if (util.safehasattr(self.data, 'next')

1126

if (util.safehasattr(self.data, 'next')

1127

or util.safehasattr(self.data, '__next__')):

1127

or util.safehasattr(self.data, '__next__')):

1128

buff = util.chunkbuffer(self.data)

1128

buff = util.chunkbuffer(self.data)

1129

chunk = buff.read(preferedchunksize)

1129

chunk = buff.read(preferedchunksize)

1130

while chunk:

1130

while chunk:

1131

yield chunk

1131

yield chunk

1132

chunk = buff.read(preferedchunksize)

1132

chunk = buff.read(preferedchunksize)

1133

elif len(self.data):

1133

elif len(self.data):

1134

yield self.data

1134

yield self.data

1135

1136

1137

flaginterrupt = -1

1137

flaginterrupt = -1

1138

1139

class interrupthandler(unpackermixin):

1139

class interrupthandler(unpackermixin):

1140

"""read one part and process it with restricted capability

1140

"""read one part and process it with restricted capability

1141

1142

This allows to transmit exception raised on the producer size during part

1142

This allows to transmit exception raised on the producer size during part

1143

iteration while the consumer is reading a part.

1143

iteration while the consumer is reading a part.

1144

1145

Part processed in this manner only have access to a ui object,"""

1145

Part processed in this manner only have access to a ui object,"""

1146

1147

def __init__(self, ui, fp):

1147

def __init__(self, ui, fp):

1148

super(interrupthandler, self).__init__(fp)

1148

super(interrupthandler, self).__init__(fp)

1149

self.ui = ui

1149

self.ui = ui

1150

1151

def _readpartheader(self):

1151

def _readpartheader(self):

1152

"""reads a part header size and return the bytes blob

1152

"""reads a part header size and return the bytes blob

1153

1154

returns None if empty"""

1154

returns None if empty"""

1155

headersize = self._unpack(_fpartheadersize)[0]

1155

headersize = self._unpack(_fpartheadersize)[0]

1156

if headersize < 0:

1156

if headersize < 0:

1157

raise error.BundleValueError('negative part header size: %i'

1157

raise error.BundleValueError('negative part header size: %i'

1158

% headersize)

1158

% headersize)

1159

indebug(self.ui, 'part header size: %i\n' % headersize)

1159

indebug(self.ui, 'part header size: %i\n' % headersize)

1160

if headersize:

1160

if headersize:

1161

return self._readexact(headersize)

1161

return self._readexact(headersize)

1162

return None

1162

return None

1163

1164

def __call__(self):

1164

def __call__(self):

1165

1166

self.ui.debug('bundle2-input-stream-interrupt:'

1166

self.ui.debug('bundle2-input-stream-interrupt:'

1167

' opening out of band context\n')

1167

' opening out of band context\n')

1168

indebug(self.ui, 'bundle2 stream interruption, looking for a part.')

1168

indebug(self.ui, 'bundle2 stream interruption, looking for a part.')

1169

headerblock = self._readpartheader()

1169

headerblock = self._readpartheader()

1170

if headerblock is None:

1170

if headerblock is None:

1171

indebug(self.ui, 'no part found during interruption.')

1171

indebug(self.ui, 'no part found during interruption.')

1172

return

1172

return

1173

part = unbundlepart(self.ui, headerblock, self._fp)

1173

part = unbundlepart(self.ui, headerblock, self._fp)

1174

op = interruptoperation(self.ui)

1174

op = interruptoperation(self.ui)

1175

hardabort = False

1175

hardabort = False

1176

try:

1176

try:

1177

_processpart(op, part)

1177

_processpart(op, part)

1178

except (SystemExit, KeyboardInterrupt):

1178

except (SystemExit, KeyboardInterrupt):

1179

hardabort = True

1179

hardabort = True

1180

raise

1180

raise

1181

finally:

1181

finally:

1182

if not hardabort:

1182

if not hardabort:

1183

part.consume()

1183

part.consume()

1184

self.ui.debug('bundle2-input-stream-interrupt:'

1184

self.ui.debug('bundle2-input-stream-interrupt:'

1185

' closing out of band context\n')

1185

' closing out of band context\n')

1186

1187

class interruptoperation(object):

1187

class interruptoperation(object):

1188

"""A limited operation to be use by part handler during interruption

1188

"""A limited operation to be use by part handler during interruption

1189

1190

It only have access to an ui object.

1190

It only have access to an ui object.

1191

"""

1191

"""

1192

1193

def __init__(self, ui):

1193

def __init__(self, ui):

1194

self.ui = ui

1194

self.ui = ui

1195

self.reply = None

1195

self.reply = None

1196

self.captureoutput = False

1196

self.captureoutput = False

1197

1198

@property

1198

@property

1199

def repo(self):

1199

def repo(self):

1200

raise error.ProgrammingError('no repo access from stream interruption')

1200

raise error.ProgrammingError('no repo access from stream interruption')

1201

1202

def gettransaction(self):

1202

def gettransaction(self):

1203

raise TransactionUnavailable('no repo access from stream interruption')

1203

raise TransactionUnavailable('no repo access from stream interruption')

1204

1205

def decodepayloadchunks(ui, fh):

1205

def decodepayloadchunks(ui, fh):

1206

"""Reads bundle2 part payload data into chunks.

1206

"""Reads bundle2 part payload data into chunks.

1207

1208

Part payload data consists of framed chunks. This function takes

1208

Part payload data consists of framed chunks. This function takes

1209

a file handle and emits those chunks.

1209

a file handle and emits those chunks.

1210

"""

1210

"""

1211

dolog = ui.configbool('devel', 'bundle2.debug')

1211

dolog = ui.configbool('devel', 'bundle2.debug')

1212

debug = ui.debug

1212

debug = ui.debug

1213

1214

headerstruct = struct.Struct(_fpayloadsize)

1214

headerstruct = struct.Struct(_fpayloadsize)

1215

headersize = headerstruct.size

1215

headersize = headerstruct.size

1216

unpack = headerstruct.unpack

1216

unpack = headerstruct.unpack

1217

1218

readexactly = changegroup.readexactly

1218

readexactly = changegroup.readexactly

1219

read = fh.read

1219

read = fh.read

1220

1221

chunksize = unpack(readexactly(fh, headersize))[0]

1221

chunksize = unpack(readexactly(fh, headersize))[0]

1222

indebug(ui, 'payload chunk size: %i' % chunksize)

1222

indebug(ui, 'payload chunk size: %i' % chunksize)

1223

1224

# changegroup.readexactly() is inlined below for performance.

1224

# changegroup.readexactly() is inlined below for performance.

1225

while chunksize:

1225

while chunksize:

1226

if chunksize >= 0:

1226

if chunksize >= 0:

1227

s = read(chunksize)

1227

s = read(chunksize)

1228

if len(s) < chunksize:

1228

if len(s) < chunksize:

1229

raise error.Abort(_('stream ended unexpectedly '

1229

raise error.Abort(_('stream ended unexpectedly '

1230

' (got %d bytes, expected %d)') %

1230

' (got %d bytes, expected %d)') %

1231

(len(s), chunksize))

1231

(len(s), chunksize))

1232

1233

yield s

1233

yield s

1234

elif chunksize == flaginterrupt:

1234

elif chunksize == flaginterrupt:

1235

# Interrupt "signal" detected. The regular stream is interrupted

1235

# Interrupt "signal" detected. The regular stream is interrupted

1236

# and a bundle2 part follows. Consume it.

1236

# and a bundle2 part follows. Consume it.

1237

interrupthandler(ui, fh)()

1237

interrupthandler(ui, fh)()

1238

else:

1238

else:

1239

raise error.BundleValueError(

1239

raise error.BundleValueError(

1240

'negative payload chunk size: %s' % chunksize)

1240

'negative payload chunk size: %s' % chunksize)

1241

1242

s = read(headersize)

1242

s = read(headersize)

1243

if len(s) < headersize:

1243

if len(s) < headersize:

1244

raise error.Abort(_('stream ended unexpectedly '

1244

raise error.Abort(_('stream ended unexpectedly '

1245

' (got %d bytes, expected %d)') %

1245

' (got %d bytes, expected %d)') %

1246

(len(s), chunksize))

1246

(len(s), chunksize))

1247

1248

chunksize = unpack(s)[0]

1248

chunksize = unpack(s)[0]

1249

1250

# indebug() inlined for performance.

1250

# indebug() inlined for performance.

1251

if dolog:

1251

if dolog:

1252

debug('bundle2-input: payload chunk size: %i\n' % chunksize)

1252

debug('bundle2-input: payload chunk size: %i\n' % chunksize)

1253

1254

class unbundlepart(unpackermixin):

1254

class unbundlepart(unpackermixin):

1255

"""a bundle part read from a bundle"""

1255

"""a bundle part read from a bundle"""

1256

1257

def __init__(self, ui, header, fp):

1257

def __init__(self, ui, header, fp):

1258

super(unbundlepart, self).__init__(fp)

1258

super(unbundlepart, self).__init__(fp)

1259

self._seekable = (util.safehasattr(fp, 'seek') and

1259

self._seekable = (util.safehasattr(fp, 'seek') and

1260

util.safehasattr(fp, 'tell'))

1260

util.safehasattr(fp, 'tell'))

1261

self.ui = ui

1261

self.ui = ui

1262

# unbundle state attr

1262

# unbundle state attr

1263

self._headerdata = header

1263

self._headerdata = header

1264

self._headeroffset = 0

1264

self._headeroffset = 0

1265

self._initialized = False

1265

self._initialized = False

1266

self.consumed = False

1266

self.consumed = False

1267

# part data

1267

# part data

1268

self.id = None

1268

self.id = None

1269

self.type = None

1269

self.type = None

1270

self.mandatoryparams = None

1270

self.mandatoryparams = None

1271

self.advisoryparams = None

1271

self.advisoryparams = None

1272

self.params = None

1272

self.params = None

1273

self.mandatorykeys = ()

1273

self.mandatorykeys = ()

1274

self._readheader()

1274

self._readheader()

1275

self._mandatory = None

1275

self._mandatory = None

1276

self._pos = 0

1276

self._pos = 0

1277

1278

def _fromheader(self, size):

1278

def _fromheader(self, size):

1279

"""return the next <size> byte from the header"""

1279

"""return the next <size> byte from the header"""

1280

offset = self._headeroffset

1280

offset = self._headeroffset

1281

data = self._headerdata[offset:(offset + size)]

1281

data = self._headerdata[offset:(offset + size)]

1282

self._headeroffset = offset + size

1282

self._headeroffset = offset + size

1283

return data

1283

return data

1284

1285

def _unpackheader(self, format):

1285

def _unpackheader(self, format):

1286

"""read given format from header

1286

"""read given format from header

1287

1288

This automatically compute the size of the format to read."""

1288

This automatically compute the size of the format to read."""

1289

data = self._fromheader(struct.calcsize(format))

1289

data = self._fromheader(struct.calcsize(format))

1290

return _unpack(format, data)

1290

return _unpack(format, data)

1291

1292

def _initparams(self, mandatoryparams, advisoryparams):

1292

def _initparams(self, mandatoryparams, advisoryparams):

1293

"""internal function to setup all logic related parameters"""

1293

"""internal function to setup all logic related parameters"""

1294

# make it read only to prevent people touching it by mistake.

1294

# make it read only to prevent people touching it by mistake.

1295

self.mandatoryparams = tuple(mandatoryparams)

1295

self.mandatoryparams = tuple(mandatoryparams)

1296

self.advisoryparams = tuple(advisoryparams)

1296

self.advisoryparams = tuple(advisoryparams)

1297

# user friendly UI

1297

# user friendly UI

1298

self.params = util.sortdict(self.mandatoryparams)

1298

self.params = util.sortdict(self.mandatoryparams)

1299

self.params.update(self.advisoryparams)

1299

self.params.update(self.advisoryparams)

1300

self.mandatorykeys = frozenset(p[0] for p in mandatoryparams)

1300

self.mandatorykeys = frozenset(p[0] for p in mandatoryparams)

1301

1302

def _readheader(self):

1302

def _readheader(self):

1303

"""read the header and setup the object"""

1303

"""read the header and setup the object"""

1304

typesize = self._unpackheader(_fparttypesize)[0]

1304

typesize = self._unpackheader(_fparttypesize)[0]

1305

self.type = self._fromheader(typesize)

1305

self.type = self._fromheader(typesize)

1306

indebug(self.ui, 'part type: "%s"' % self.type)

1306

indebug(self.ui, 'part type: "%s"' % self.type)

1307

self.id = self._unpackheader(_fpartid)[0]

1307

self.id = self._unpackheader(_fpartid)[0]

1308

indebug(self.ui, 'part id: "%s"' % pycompat.bytestr(self.id))

1308

indebug(self.ui, 'part id: "%s"' % pycompat.bytestr(self.id))

1309

# extract mandatory bit from type

1309

# extract mandatory bit from type

1310

self.mandatory = (self.type != self.type.lower())

1310

self.mandatory = (self.type != self.type.lower())

1311

self.type = self.type.lower()

1311

self.type = self.type.lower()

1312

## reading parameters

1312

## reading parameters

1313

# param count

1313

# param count

1314

mancount, advcount = self._unpackheader(_fpartparamcount)

1314

mancount, advcount = self._unpackheader(_fpartparamcount)

1315

indebug(self.ui, 'part parameters: %i' % (mancount + advcount))

1315

indebug(self.ui, 'part parameters: %i' % (mancount + advcount))

1316

# param size

1316

# param size

1317

fparamsizes = _makefpartparamsizes(mancount + advcount)

1317

fparamsizes = _makefpartparamsizes(mancount + advcount)

1318

paramsizes = self._unpackheader(fparamsizes)

1318

paramsizes = self._unpackheader(fparamsizes)

1319

# make it a list of couple again

1319

# make it a list of couple again

1320

paramsizes = list(zip(paramsizes[::2], paramsizes[1::2]))

1320

paramsizes = list(zip(paramsizes[::2], paramsizes[1::2]))

1321

# split mandatory from advisory

1321

# split mandatory from advisory

1322

mansizes = paramsizes[:mancount]

1322

mansizes = paramsizes[:mancount]

1323

advsizes = paramsizes[mancount:]

1323

advsizes = paramsizes[mancount:]

1324

# retrieve param value

1324

# retrieve param value

1325

manparams = []

1325

manparams = []

1326

for key, value in mansizes:

1326

for key, value in mansizes:

1327

manparams.append((self._fromheader(key), self._fromheader(value)))

1327

manparams.append((self._fromheader(key), self._fromheader(value)))

1328

advparams = []

1328

advparams = []

1329

for key, value in advsizes:

1329

for key, value in advsizes:

1330

advparams.append((self._fromheader(key), self._fromheader(value)))

1330

advparams.append((self._fromheader(key), self._fromheader(value)))

1331

self._initparams(manparams, advparams)

1331

self._initparams(manparams, advparams)

1332

## part payload

1332

## part payload

1333

self._payloadstream = util.chunkbuffer(self._payloadchunks())

1333

self._payloadstream = util.chunkbuffer(self._payloadchunks())

1334

# we read the data, tell it

1334

# we read the data, tell it

1335

self._initialized = True

1335

self._initialized = True

1336

1337

def _payloadchunks(self):

1337

def _payloadchunks(self):

1338

"""Generator of decoded chunks in the payload."""

1338

"""Generator of decoded chunks in the payload."""

1339

return decodepayloadchunks(self.ui, self._fp)

1339

return decodepayloadchunks(self.ui, self._fp)

1340

1341

def consume(self):

1341

def consume(self):

1342

"""Read the part payload until completion.

1342

"""Read the part payload until completion.

1343

1344

By consuming the part data, the underlying stream read offset will

1344

By consuming the part data, the underlying stream read offset will

1345

be advanced to the next part (or end of stream).

1345

be advanced to the next part (or end of stream).

1346

"""

1346

"""

1347

if self.consumed:

1347

if self.consumed:

1348

return

1348

return

1349

1350

chunk = self.read(32768)

1350

chunk = self.read(32768)

1351

while chunk:

1351

while chunk:

1352

self._pos += len(chunk)

1352

self._pos += len(chunk)

1353

chunk = self.read(32768)

1353

chunk = self.read(32768)

1354

1355

def read(self, size=None):

1355

def read(self, size=None):

1356

"""read payload data"""

1356

"""read payload data"""

1357

if not self._initialized:

1357

if not self._initialized:

1358

self._readheader()

1358

self._readheader()

1359

if size is None:

1359

if size is None:

1360

data = self._payloadstream.read()

1360

data = self._payloadstream.read()

1361

else:

1361

else:

1362

data = self._payloadstream.read(size)

1362

data = self._payloadstream.read(size)

1363

self._pos += len(data)

1363

self._pos += len(data)

1364

if size is None or len(data) < size:

1364

if size is None or len(data) < size:

1365

if not self.consumed and self._pos:

1365

if not self.consumed and self._pos:

1366

self.ui.debug('bundle2-input-part: total payload size %i\n'

1366

self.ui.debug('bundle2-input-part: total payload size %i\n'

1367

% self._pos)

1367

% self._pos)

1368

self.consumed = True

1368

self.consumed = True

1369

return data

1369

return data

1370

1371

class seekableunbundlepart(unbundlepart):

1371

class seekableunbundlepart(unbundlepart):

1372

"""A bundle2 part in a bundle that is seekable.

1372

"""A bundle2 part in a bundle that is seekable.

1373

1374

Regular ``unbundlepart`` instances can only be read once. This class

1374

Regular ``unbundlepart`` instances can only be read once. This class

1375

extends ``unbundlepart`` to enable bi-directional seeking within the

1375

extends ``unbundlepart`` to enable bi-directional seeking within the

1376

part.

1376

part.

1377

1378

Bundle2 part data consists of framed chunks. Offsets when seeking

1378

Bundle2 part data consists of framed chunks. Offsets when seeking

1379

refer to the decoded data, not the offsets in the underlying bundle2

1379

refer to the decoded data, not the offsets in the underlying bundle2

1380

stream.

1380

stream.

1381

1382

To facilitate quickly seeking within the decoded data, instances of this

1382

To facilitate quickly seeking within the decoded data, instances of this

1383

class maintain a mapping between offsets in the underlying stream and

1383

class maintain a mapping between offsets in the underlying stream and

1384

the decoded payload. This mapping will consume memory in proportion

1384

the decoded payload. This mapping will consume memory in proportion

1385

to the number of chunks within the payload (which almost certainly

1385

to the number of chunks within the payload (which almost certainly

1386

increases in proportion with the size of the part).

1386

increases in proportion with the size of the part).

1387

"""

1387

"""

1388

def __init__(self, ui, header, fp):

1388

def __init__(self, ui, header, fp):

1389

# (payload, file) offsets for chunk starts.

1389

# (payload, file) offsets for chunk starts.

1390

self._chunkindex = []

1390

self._chunkindex = []

1391

1392

super(seekableunbundlepart, self).__init__(ui, header, fp)

1392

super(seekableunbundlepart, self).__init__(ui, header, fp)

1393

1394

def _payloadchunks(self, chunknum=0):

1394

def _payloadchunks(self, chunknum=0):

1395

'''seek to specified chunk and start yielding data'''

1395

'''seek to specified chunk and start yielding data'''

1396

if len(self._chunkindex) == 0:

1396

if len(self._chunkindex) == 0:

1397

assert chunknum == 0, 'Must start with chunk 0'

1397

assert chunknum == 0, 'Must start with chunk 0'

1398

self._chunkindex.append((0, self._tellfp()))

1398

self._chunkindex.append((0, self._tellfp()))

1399

else:

1399

else:

1400

assert chunknum < len(self._chunkindex), \

1400

assert chunknum < len(self._chunkindex), \

1401

'Unknown chunk %d' % chunknum

1401

'Unknown chunk %d' % chunknum

1402

self._seekfp(self._chunkindex[chunknum][1])

1402

self._seekfp(self._chunkindex[chunknum][1])

1403

1404

pos = self._chunkindex[chunknum][0]

1404

pos = self._chunkindex[chunknum][0]

1405

1406

for chunk in decodepayloadchunks(self.ui, self._fp):

1406

for chunk in decodepayloadchunks(self.ui, self._fp):

1407

chunknum += 1

1407

chunknum += 1

1408

pos += len(chunk)

1408

pos += len(chunk)

1409

if chunknum == len(self._chunkindex):

1409

if chunknum == len(self._chunkindex):

1410

self._chunkindex.append((pos, self._tellfp()))

1410

self._chunkindex.append((pos, self._tellfp()))

1411

1412

yield chunk

1412

yield chunk

1413

1414

def _findchunk(self, pos):

1414

def _findchunk(self, pos):

1415

'''for a given payload position, return a chunk number and offset'''

1415

'''for a given payload position, return a chunk number and offset'''

1416

for chunk, (ppos, fpos) in enumerate(self._chunkindex):

1416

for chunk, (ppos, fpos) in enumerate(self._chunkindex):

1417

if ppos == pos:

1417

if ppos == pos:

1418

return chunk, 0

1418

return chunk, 0

1419

elif ppos > pos:

1419

elif ppos > pos:

1420

return chunk - 1, pos - self._chunkindex[chunk - 1][0]

1420

return chunk - 1, pos - self._chunkindex[chunk - 1][0]

1421

raise ValueError('Unknown chunk')

1421

raise ValueError('Unknown chunk')

1422

1423

def tell(self):

1423

def tell(self):

1424

return self._pos

1424

return self._pos

1425

1426

def seek(self, offset, whence=os.SEEK_SET):

1426

def seek(self, offset, whence=os.SEEK_SET):

1427

if whence == os.SEEK_SET:

1427

if whence == os.SEEK_SET:

1428

newpos = offset

1428

newpos = offset

1429

elif whence == os.SEEK_CUR:

1429

elif whence == os.SEEK_CUR:

1430

newpos = self._pos + offset

1430

newpos = self._pos + offset

1431

elif whence == os.SEEK_END:

1431

elif whence == os.SEEK_END:

1432

if not self.consumed:

1432

if not self.consumed:

1433

# Can't use self.consume() here because it advances self._pos.

1433

# Can't use self.consume() here because it advances self._pos.

1434

chunk = self.read(32768)

1434

chunk = self.read(32768)

1435

while chunk:

1435

while chunk:

1436

chunk = self.read(32768)

1436

chunk = self.read(32768)

1437

newpos = self._chunkindex[-1][0] - offset

1437

newpos = self._chunkindex[-1][0] - offset

1438

else:

1438

else:

1439

raise ValueError('Unknown whence value: %r' % (whence,))

1439

raise ValueError('Unknown whence value: %r' % (whence,))

1440

1441

if newpos > self._chunkindex[-1][0] and not self.consumed:

1441

if newpos > self._chunkindex[-1][0] and not self.consumed:

1442

# Can't use self.consume() here because it advances self._pos.

1442

# Can't use self.consume() here because it advances self._pos.

1443

chunk = self.read(32768)

1443

chunk = self.read(32768)

1444

while chunk:

1444

while chunk:

1445

chunk = self.read(32668)

1445

chunk = self.read(32668)

1446

1447

if not 0 <= newpos <= self._chunkindex[-1][0]:

1447

if not 0 <= newpos <= self._chunkindex[-1][0]:

1448

raise ValueError('Offset out of range')

1448

raise ValueError('Offset out of range')

1449

1450

if self._pos != newpos:

1450

if self._pos != newpos:

1451

chunk, internaloffset = self._findchunk(newpos)

1451

chunk, internaloffset = self._findchunk(newpos)

1452

self._payloadstream = util.chunkbuffer(self._payloadchunks(chunk))

1452

self._payloadstream = util.chunkbuffer(self._payloadchunks(chunk))

1453

adjust = self.read(internaloffset)

1453

adjust = self.read(internaloffset)

1454

if len(adjust) != internaloffset:

1454

if len(adjust) != internaloffset:

1455

raise error.Abort(_('Seek failed\n'))

1455

raise error.Abort(_('Seek failed\n'))

1456

self._pos = newpos

1456

self._pos = newpos

1457

1458

def _seekfp(self, offset, whence=0):

1458

def _seekfp(self, offset, whence=0):

1459

"""move the underlying file pointer

1459

"""move the underlying file pointer

1460

1461

This method is meant for internal usage by the bundle2 protocol only.

1461

This method is meant for internal usage by the bundle2 protocol only.

1462

They directly manipulate the low level stream including bundle2 level

1462

They directly manipulate the low level stream including bundle2 level

1463

instruction.

1463

instruction.

1464

1465

Do not use it to implement higher-level logic or methods."""

1465

Do not use it to implement higher-level logic or methods."""

1466

if self._seekable:

1466

if self._seekable:

1467

return self._fp.seek(offset, whence)

1467

return self._fp.seek(offset, whence)

1468

else:

1468

else:

1469

raise NotImplementedError(_('File pointer is not seekable'))

1469

raise NotImplementedError(_('File pointer is not seekable'))

1470

1471

def _tellfp(self):

1471

def _tellfp(self):

1472

"""return the file offset, or None if file is not seekable

1472

"""return the file offset, or None if file is not seekable

1473

1474

This method is meant for internal usage by the bundle2 protocol only.

1474

This method is meant for internal usage by the bundle2 protocol only.

1475

They directly manipulate the low level stream including bundle2 level

1475

They directly manipulate the low level stream including bundle2 level

1476

instruction.

1476

instruction.

1477

1478

Do not use it to implement higher-level logic or methods."""

1478

Do not use it to implement higher-level logic or methods."""

1479

if self._seekable:

1479

if self._seekable:

1480

try:

1480

try:

1481

return self._fp.tell()

1481

return self._fp.tell()

1482

except IOError as e:

1482

except IOError as e:

1483

if e.errno == errno.ESPIPE:

1483

if e.errno == errno.ESPIPE:

1484

self._seekable = False

1484

self._seekable = False

1485

else:

1485

else:

1486

raise

1486

raise

1487

return None

1487

return None

1488

1489

# These are only the static capabilities.

1489

# These are only the static capabilities.

1490

# Check the 'getrepocaps' function for the rest.

1490

# Check the 'getrepocaps' function for the rest.

1491

capabilities = {'HG20': (),

1491

capabilities = {'HG20': (),

1492

'bookmarks': (),

1492

'bookmarks': (),

1493

'error': ('abort', 'unsupportedcontent', 'pushraced',

1493

'error': ('abort', 'unsupportedcontent', 'pushraced',

1494

'pushkey'),

1494

'pushkey'),

1495

'listkeys': (),

1495

'listkeys': (),

1496

'pushkey': (),

1496

'pushkey': (),

1497

'digests': tuple(sorted(util.DIGESTS.keys())),

1497

'digests': tuple(sorted(util.DIGESTS.keys())),

1498

'remote-changegroup': ('http', 'https'),

1498

'remote-changegroup': ('http', 'https'),

1499

'hgtagsfnodes': (),

1499

'hgtagsfnodes': (),

1500

'rev-branch-cache': (),

1500

'rev-branch-cache': (),

1501

'phases': ('heads',),

1501

'phases': ('heads',),

1502

'stream': ('v2',),

1502

'stream': ('v2',),

1503

}

1503

}

1504

1505

def getrepocaps(repo, allowpushback=False, role=None):

1505

def getrepocaps(repo, allowpushback=False, role=None):

1506

"""return the bundle2 capabilities for a given repo

1506

"""return the bundle2 capabilities for a given repo

1507

1508

Exists to allow extensions (like evolution) to mutate the capabilities.

1508

Exists to allow extensions (like evolution) to mutate the capabilities.

1509

1510

The returned value is used for servers advertising their capabilities as

1510

The returned value is used for servers advertising their capabilities as

1511

well as clients advertising their capabilities to servers as part of

1511

well as clients advertising their capabilities to servers as part of

1512

bundle2 requests. The ``role`` argument specifies which is which.

1512

bundle2 requests. The ``role`` argument specifies which is which.

1513

"""

1513

"""

1514

if role not in ('client', 'server'):

1514

if role not in ('client', 'server'):

1515

raise error.ProgrammingError('role argument must be client or server')

1515

raise error.ProgrammingError('role argument must be client or server')

1516

1517

caps = capabilities.copy()

1517

caps = capabilities.copy()

1518

caps['changegroup'] = tuple(sorted(

1518

caps['changegroup'] = tuple(sorted(

1519

changegroup.supportedincomingversions(repo)))

1519

changegroup.supportedincomingversions(repo)))

1520

if obsolete.isenabled(repo, obsolete.exchangeopt):

1520

if obsolete.isenabled(repo, obsolete.exchangeopt):

1521

supportedformat = tuple('V%i' % v for v in obsolete.formats)

1521

supportedformat = tuple('V%i' % v for v in obsolete.formats)

1522

caps['obsmarkers'] = supportedformat

1522

caps['obsmarkers'] = supportedformat

1523

if allowpushback:

1523

if allowpushback:

1524

caps['pushback'] = ()

1524

caps['pushback'] = ()

1525

cpmode = repo.ui.config('server', 'concurrent-push-mode')

1525

cpmode = repo.ui.config('server', 'concurrent-push-mode')

1526

if cpmode == 'check-related':

1526

if cpmode == 'check-related':

1527

caps['checkheads'] = ('related',)

1527

caps['checkheads'] = ('related',)

1528

if 'phases' in repo.ui.configlist('devel', 'legacy.exchange'):

1528

if 'phases' in repo.ui.configlist('devel', 'legacy.exchange'):

1529

caps.pop('phases')

1529

caps.pop('phases')

1530

1531

# Don't advertise stream clone support in server mode if not configured.

1531

# Don't advertise stream clone support in server mode if not configured.

1532

if role == 'server':

1532

if role == 'server':

1533

streamsupported = repo.ui.configbool('server', 'uncompressed',

1533

streamsupported = repo.ui.configbool('server', 'uncompressed',

1534

untrusted=True)

1534

untrusted=True)

1535

featuresupported = repo.ui.configbool('experimental', 'bundle2.stream')

1535

featuresupported = repo.ui.configbool('experimental', 'bundle2.stream')

1536

1537

if not streamsupported or not featuresupported:

1537

if not streamsupported or not featuresupported:

1538

caps.pop('stream')

1538

caps.pop('stream')

1539

# Else always advertise support on client, because payload support

1539

# Else always advertise support on client, because payload support

1540

# should always be advertised.

1540

# should always be advertised.

1541

1542

return caps

1542

return caps

1543

1544

def bundle2caps(remote):

1544

def bundle2caps(remote):

1545

"""return the bundle capabilities of a peer as dict"""

1545

"""return the bundle capabilities of a peer as dict"""

1546

raw = remote.capable('bundle2')

1546

raw = remote.capable('bundle2')

1547

if not raw and raw != '':

1547

if not raw and raw != '':

1548

return {}

1548

return {}

1549

capsblob = urlreq.unquote(remote.capable('bundle2'))

1549

capsblob = urlreq.unquote(remote.capable('bundle2'))

1550

return decodecaps(capsblob)

1550

return decodecaps(capsblob)

1551

1552

def obsmarkersversion(caps):

1552

def obsmarkersversion(caps):

1553

"""extract the list of supported obsmarkers versions from a bundle2caps dict

1553

"""extract the list of supported obsmarkers versions from a bundle2caps dict

1554

"""

1554

"""

1555

obscaps = caps.get('obsmarkers', ())

1555

obscaps = caps.get('obsmarkers', ())

1556

return [int(c[1:]) for c in obscaps if c.startswith('V')]

1556

return [int(c[1:]) for c in obscaps if c.startswith('V')]

1557

1558

def writenewbundle(ui, repo, source, filename, bundletype, outgoing, opts,

1558

def writenewbundle(ui, repo, source, filename, bundletype, outgoing, opts,

1559

vfs=None, compression=None, compopts=None):

1559

vfs=None, compression=None, compopts=None):

1560

if bundletype.startswith('HG10'):

1560

if bundletype.startswith('HG10'):

1561

cg = changegroup.makechangegroup(repo, outgoing, '01', source)

1561

cg = changegroup.makechangegroup(repo, outgoing, '01', source)

1562

return writebundle(ui, cg, filename, bundletype, vfs=vfs,

1562

return writebundle(ui, cg, filename, bundletype, vfs=vfs,

1563

compression=compression, compopts=compopts)

1563

compression=compression, compopts=compopts)

1564

elif not bundletype.startswith('HG20'):

1564

elif not bundletype.startswith('HG20'):

1565

raise error.ProgrammingError('unknown bundle type: %s' % bundletype)

1565

raise error.ProgrammingError('unknown bundle type: %s' % bundletype)

1566

1567

caps = {}

1567

caps = {}

1568

if 'obsolescence' in opts:

1568

if 'obsolescence' in opts:

1569

caps['obsmarkers'] = ('V1',)

1569

caps['obsmarkers'] = ('V1',)

1570

bundle = bundle20(ui, caps)

1570

bundle = bundle20(ui, caps)

1571

bundle.setcompression(compression, compopts)

1571

bundle.setcompression(compression, compopts)

1572

_addpartsfromopts(ui, repo, bundle, source, outgoing, opts)

1572

_addpartsfromopts(ui, repo, bundle, source, outgoing, opts)

1573

chunkiter = bundle.getchunks()

1573

chunkiter = bundle.getchunks()

1574

1575

return changegroup.writechunks(ui, chunkiter, filename, vfs=vfs)

1575

return changegroup.writechunks(ui, chunkiter, filename, vfs=vfs)

1576

1577

def _addpartsfromopts(ui, repo, bundler, source, outgoing, opts):

1577

def _addpartsfromopts(ui, repo, bundler, source, outgoing, opts):

1578

# We should eventually reconcile this logic with the one behind

1578

# We should eventually reconcile this logic with the one behind

1579

# 'exchange.getbundle2partsgenerator'.

1579

# 'exchange.getbundle2partsgenerator'.

1580

#

1580

#

1581

# The type of input from 'getbundle' and 'writenewbundle' are a bit

1581

# The type of input from 'getbundle' and 'writenewbundle' are a bit

1582

# different right now. So we keep them separated for now for the sake of

1582

# different right now. So we keep them separated for now for the sake of

1583

# simplicity.

1583

# simplicity.

1584

1585

# we might not always want a changegroup in such bundle, for example in

1585

# we might not always want a changegroup in such bundle, for example in

1586

# stream bundles

1586

# stream bundles

1587

if opts.get('changegroup', True):

1587

if opts.get('changegroup', True):

1588

cgversion = opts.get('cg.version')

1588

cgversion = opts.get('cg.version')

1589

if cgversion is None:

1589

if cgversion is None:

1590

cgversion = changegroup.safeversion(repo)

1590

cgversion = changegroup.safeversion(repo)

1591

cg = changegroup.makechangegroup(repo, outgoing, cgversion, source)

1591

cg = changegroup.makechangegroup(repo, outgoing, cgversion, source)

1592

part = bundler.newpart('changegroup', data=cg.getchunks())

1592

part = bundler.newpart('changegroup', data=cg.getchunks())

1593

part.addparam('version', cg.version)

1593

part.addparam('version', cg.version)

1594

if 'clcount' in cg.extras:

1594

if 'clcount' in cg.extras:

1595

part.addparam('nbchanges', '%d' % cg.extras['clcount'],

1595

part.addparam('nbchanges', '%d' % cg.extras['clcount'],

1596

mandatory=False)

1596

mandatory=False)

1597

if opts.get('phases') and repo.revs('%ln and secret()',

1597

if opts.get('phases') and repo.revs('%ln and secret()',

1598

outgoing.missingheads):

1598

outgoing.missingheads):

1599

part.addparam('targetphase', '%d' % phases.secret, mandatory=False)

1599

part.addparam('targetphase', '%d' % phases.secret, mandatory=False)

1600

1601

if opts.get('streamv2', False):

1601

if opts.get('streamv2', False):

1602

addpartbundlestream2(bundler, repo, stream=True)

1602

addpartbundlestream2(bundler, repo, stream=True)

1603

1604

if opts.get('tagsfnodescache', True):

1604

if opts.get('tagsfnodescache', True):

1605

addparttagsfnodescache(repo, bundler, outgoing)

1605

addparttagsfnodescache(repo, bundler, outgoing)

1606

1607

if opts.get('revbranchcache', True):

1607

if opts.get('revbranchcache', True):

1608

addpartrevbranchcache(repo, bundler, outgoing)

1608

addpartrevbranchcache(repo, bundler, outgoing)

1609

1610

if opts.get('obsolescence', False):

1610

if opts.get('obsolescence', False):

1611

obsmarkers = repo.obsstore.relevantmarkers(outgoing.missing)

1611

obsmarkers = repo.obsstore.relevantmarkers(outgoing.missing)

1612

buildobsmarkerspart(bundler, obsmarkers)

1612

buildobsmarkerspart(bundler, obsmarkers)

1613

1614

if opts.get('phases', False):

1614

if opts.get('phases', False):

1615

headsbyphase = phases.subsetphaseheads(repo, outgoing.missing)

1615

headsbyphase = phases.subsetphaseheads(repo, outgoing.missing)

1616

phasedata = phases.binaryencode(headsbyphase)

1616

phasedata = phases.binaryencode(headsbyphase)

1617

bundler.newpart('phase-heads', data=phasedata)

1617

bundler.newpart('phase-heads', data=phasedata)

1618

1619

def addparttagsfnodescache(repo, bundler, outgoing):

1619

def addparttagsfnodescache(repo, bundler, outgoing):

1620

# we include the tags fnode cache for the bundle changeset

1620

# we include the tags fnode cache for the bundle changeset

1621

# (as an optional parts)

1621

# (as an optional parts)

1622

cache = tags.hgtagsfnodescache(repo.unfiltered())

1622

cache = tags.hgtagsfnodescache(repo.unfiltered())

1623

chunks = []

1623

chunks = []

1624

1625

# .hgtags fnodes are only relevant for head changesets. While we could

1625

# .hgtags fnodes are only relevant for head changesets. While we could

1626

# transfer values for all known nodes, there will likely be little to

1626

# transfer values for all known nodes, there will likely be little to

1627

# no benefit.

1627

# no benefit.

1628

#

1628

#

1629

# We don't bother using a generator to produce output data because

1629

# We don't bother using a generator to produce output data because

1630

# a) we only have 40 bytes per head and even esoteric numbers of heads

1630

# a) we only have 40 bytes per head and even esoteric numbers of heads

1631

# consume little memory (1M heads is 40MB) b) we don't want to send the

1631

# consume little memory (1M heads is 40MB) b) we don't want to send the

1632

# part if we don't have entries and knowing if we have entries requires

1632

# part if we don't have entries and knowing if we have entries requires

1633

# cache lookups.

1633

# cache lookups.

1634

for node in outgoing.missingheads:

1634

for node in outgoing.missingheads:

1635

# Don't compute missing, as this may slow down serving.

1635

# Don't compute missing, as this may slow down serving.

1636

fnode = cache.getfnode(node, computemissing=False)

1636

fnode = cache.getfnode(node, computemissing=False)

1637

if fnode is not None:

1637

if fnode is not None:

1638

chunks.extend([node, fnode])

1638

chunks.extend([node, fnode])

1639

1640

if chunks:

1640

if chunks:

1641

bundler.newpart('hgtagsfnodes', data=''.join(chunks))

1641

bundler.newpart('hgtagsfnodes', data=''.join(chunks))

1642

1643

def addpartrevbranchcache(repo, bundler, outgoing):

1643

def addpartrevbranchcache(repo, bundler, outgoing):

1644

# we include the rev branch cache for the bundle changeset

1644

# we include the rev branch cache for the bundle changeset

1645

# (as an optional parts)

1645

# (as an optional parts)

1646

cache = repo.revbranchcache()

1646

cache = repo.revbranchcache()

1647

cl = repo.unfiltered().changelog

1647

cl = repo.unfiltered().changelog

1648

branchesdata = collections.defaultdict(lambda: (set(), set()))

1648

branchesdata = collections.defaultdict(lambda: (set(), set()))

1649

for node in outgoing.missing:

1649

for node in outgoing.missing:

1650

branch, close = cache.branchinfo(cl.rev(node))

1650

branch, close = cache.branchinfo(cl.rev(node))

1651

branchesdata[branch][close].add(node)

1651

branchesdata[branch][close].add(node)

1652

1653

def generate():

1653

def generate():

1654

for branch, (nodes, closed) in sorted(branchesdata.items()):

1654

for branch, (nodes, closed) in sorted(branchesdata.items()):

1655

utf8branch = encoding.fromlocal(branch)

1655

utf8branch = encoding.fromlocal(branch)

1656

yield rbcstruct.pack(len(utf8branch), len(nodes), len(closed))

1656

yield rbcstruct.pack(len(utf8branch), len(nodes), len(closed))

1657

yield utf8branch

1657

yield utf8branch

1658

for n in sorted(nodes):

1658

for n in sorted(nodes):

1659

yield n

1659

yield n

1660

for n in sorted(closed):

1660

for n in sorted(closed):

1661

yield n

1661

yield n

1662

1663

bundler.newpart('cache:rev-branch-cache', data=generate(),

1663

bundler.newpart('cache:rev-branch-cache', data=generate(),

1664

mandatory=False)

1664

mandatory=False)

1665

1666

def _formatrequirementsspec(requirements):

1666

def _formatrequirementsspec(requirements):

1667

return urlreq.quote(','.join(sorted(requirements)))

1667

return urlreq.quote(','.join(sorted(requirements)))

1668

1669

def _formatrequirementsparams(requirements):

1669

def _formatrequirementsparams(requirements):

1670

requirements = _formatrequirementsspec(requirements)

1670

requirements = _formatrequirementsspec(requirements)

1671

params = "%s%s" % (urlreq.quote("requirements="), requirements)

1671

params = "%s%s" % (urlreq.quote("requirements="), requirements)

1672

return params

1672

return params

1673

1674

def addpartbundlestream2(bundler, repo, **kwargs):

1674

def addpartbundlestream2(bundler, repo, **kwargs):

1675

if not kwargs.get('stream', False):

1675

if not kwargs.get('stream', False):

1676

return

1676

return

1677

1678

if not streamclone.allowservergeneration(repo):

1678

if not streamclone.allowservergeneration(repo):

1679

raise error.Abort(_('stream data requested but server does not allow '

1679

raise error.Abort(_('stream data requested but server does not allow '

1680

'this feature'),

1680

'this feature'),

1681

hint=_('well-behaved clients should not be '

1681

hint=_('well-behaved clients should not be '

1682

'requesting stream data from servers not '

1682

'requesting stream data from servers not '

1683

'advertising it; the client may be buggy'))

1683

'advertising it; the client may be buggy'))

1684

1685

# Stream clones don't compress well. And compression undermines a

1685

# Stream clones don't compress well. And compression undermines a

1686

# goal of stream clones, which is to be fast. Communicate the desire

1686

# goal of stream clones, which is to be fast. Communicate the desire

1687

# to avoid compression to consumers of the bundle.

1687

# to avoid compression to consumers of the bundle.

1688

bundler.prefercompressed = False

1688

bundler.prefercompressed = False

1689

1690

filecount, bytecount, it = streamclone.generatev2(repo)

1690

filecount, bytecount, it = streamclone.generatev2(repo)

1691

requirements = _formatrequirementsspec(repo.requirements)

1691

requirements = _formatrequirementsspec(repo.requirements)

1692

part = bundler.newpart('stream2', data=it)

1692

part = bundler.newpart('stream2', data=it)

1693

part.addparam('bytecount', '%d' % bytecount, mandatory=True)

1693

part.addparam('bytecount', '%d' % bytecount, mandatory=True)

1694

part.addparam('filecount', '%d' % filecount, mandatory=True)

1694

part.addparam('filecount', '%d' % filecount, mandatory=True)

1695

part.addparam('requirements', requirements, mandatory=True)

1695

part.addparam('requirements', requirements, mandatory=True)

1696

1697

def buildobsmarkerspart(bundler, markers):

1697

def buildobsmarkerspart(bundler, markers):

1698

"""add an obsmarker part to the bundler with <markers>

1698

"""add an obsmarker part to the bundler with <markers>

1699

1700

No part is created if markers is empty.

1700

No part is created if markers is empty.

1701

Raises ValueError if the bundler doesn't support any known obsmarker format.

1701

Raises ValueError if the bundler doesn't support any known obsmarker format.

1702

"""

1702

"""

1703

if not markers:

1703

if not markers:

1704

return None

1704

return None

1705

1706

remoteversions = obsmarkersversion(bundler.capabilities)

1706

remoteversions = obsmarkersversion(bundler.capabilities)

1707

version = obsolete.commonversion(remoteversions)

1707

version = obsolete.commonversion(remoteversions)

1708

if version is None:

1708

if version is None:

1709

raise ValueError('bundler does not support common obsmarker format')

1709

raise ValueError('bundler does not support common obsmarker format')

1710

stream = obsolete.encodemarkers(markers, True, version=version)

1710

stream = obsolete.encodemarkers(markers, True, version=version)

1711

return bundler.newpart('obsmarkers', data=stream)

1711

return bundler.newpart('obsmarkers', data=stream)

1712

1713

def writebundle(ui, cg, filename, bundletype, vfs=None, compression=None,

1713

def writebundle(ui, cg, filename, bundletype, vfs=None, compression=None,

1714

compopts=None):

1714

compopts=None):

1715

"""Write a bundle file and return its filename.

1715

"""Write a bundle file and return its filename.

1716

1717

Existing files will not be overwritten.

1717

Existing files will not be overwritten.

1718

If no filename is specified, a temporary file is created.

1718

If no filename is specified, a temporary file is created.

1719

bz2 compression can be turned off.

1719

bz2 compression can be turned off.

1720

The bundle file will be deleted in case of errors.

1720

The bundle file will be deleted in case of errors.

1721

"""

1721

"""

1722

1723

if bundletype == "HG20":

1723

if bundletype == "HG20":

1724

bundle = bundle20(ui)

1724

bundle = bundle20(ui)

1725

bundle.setcompression(compression, compopts)

1725

bundle.setcompression(compression, compopts)

1726

part = bundle.newpart('changegroup', data=cg.getchunks())

1726

part = bundle.newpart('changegroup', data=cg.getchunks())

1727

part.addparam('version', cg.version)

1727

part.addparam('version', cg.version)

1728

if 'clcount' in cg.extras:

1728

if 'clcount' in cg.extras:

1729

part.addparam('nbchanges', '%d' % cg.extras['clcount'],

1729

part.addparam('nbchanges', '%d' % cg.extras['clcount'],

1730

mandatory=False)

1730

mandatory=False)

1731

chunkiter = bundle.getchunks()

1731

chunkiter = bundle.getchunks()

1732

else:

1732

else:

1733

# compression argument is only for the bundle2 case

1733

# compression argument is only for the bundle2 case

1734

assert compression is None

1734

assert compression is None

1735

if cg.version != '01':

1735

if cg.version != '01':

1736

raise error.Abort(_('old bundle types only supports v1 '

1736

raise error.Abort(_('old bundle types only supports v1 '

1737

'changegroups'))

1737

'changegroups'))

1738

header, comp = bundletypes[bundletype]

1738

header, comp = bundletypes[bundletype]

1739

if comp not in util.compengines.supportedbundletypes:

1739

if comp not in util.compengines.supportedbundletypes:

1740

raise error.Abort(_('unknown stream compression type: %s')

1740

raise error.Abort(_('unknown stream compression type: %s')

1741

% comp)

1741

% comp)

1742

compengine = util.compengines.forbundletype(comp)

1742

compengine = util.compengines.forbundletype(comp)

1743

def chunkiter():

1743

def chunkiter():

1744

yield header

1744

yield header

1745

for chunk in compengine.compressstream(cg.getchunks(), compopts):

1745

for chunk in compengine.compressstream(cg.getchunks(), compopts):

1746

yield chunk

1746

yield chunk

1747

chunkiter = chunkiter()

1747

chunkiter = chunkiter()

1748

1749

# parse the changegroup data, otherwise we will block

1749

# parse the changegroup data, otherwise we will block

1750

# in case of sshrepo because we don't know the end of the stream

1750

# in case of sshrepo because we don't know the end of the stream

1751

return changegroup.writechunks(ui, chunkiter, filename, vfs=vfs)

1751

return changegroup.writechunks(ui, chunkiter, filename, vfs=vfs)

1752

1753

def combinechangegroupresults(op):

1753

def combinechangegroupresults(op):

1754

"""logic to combine 0 or more addchangegroup results into one"""

1754

"""logic to combine 0 or more addchangegroup results into one"""

1755

results = [r.get('return', 0)

1755

results = [r.get('return', 0)

1756

for r in op.records['changegroup']]

1756

for r in op.records['changegroup']]

1757

changedheads = 0

1757

changedheads = 0

1758

result = 1

1758

result = 1

1759

for ret in results:

1759

for ret in results:

1760

# If any changegroup result is 0, return 0

1760

# If any changegroup result is 0, return 0

1761

if ret == 0:

1761

if ret == 0:

1762

result = 0

1762

result = 0

1763

break

1763

break

1764

if ret < -1:

1764

if ret < -1:

1765

changedheads += ret + 1

1765

changedheads += ret + 1

1766

elif ret > 1:

1766

elif ret > 1:

1767

changedheads += ret - 1

1767

changedheads += ret - 1

1768

if changedheads > 0:

1768

if changedheads > 0:

1769

result = 1 + changedheads

1769

result = 1 + changedheads

1770

elif changedheads < 0:

1770

elif changedheads < 0:

1771

result = -1 + changedheads

1771

result = -1 + changedheads

1772

return result

1772

return result

1773

1774

@parthandler('changegroup', ('version', 'nbchanges', 'treemanifest',

1774

@parthandler('changegroup', ('version', 'nbchanges', 'treemanifest',

1775

'targetphase'))

1775

'targetphase'))

1776

def handlechangegroup(op, inpart):

1776

def handlechangegroup(op, inpart):

1777

"""apply a changegroup part on the repo

1777

"""apply a changegroup part on the repo

1778

1779

This is a very early implementation that will massive rework before being

1779

This is a very early implementation that will massive rework before being

1780

inflicted to any end-user.

1780

inflicted to any end-user.

1781

"""

1781

"""

1782

tr = op.gettransaction()

1782

tr = op.gettransaction()

1783

unpackerversion = inpart.params.get('version', '01')

1783

unpackerversion = inpart.params.get('version', '01')

1784

# We should raise an appropriate exception here

1784

# We should raise an appropriate exception here

1785

cg = changegroup.getunbundler(unpackerversion, inpart, None)

1785

cg = changegroup.getunbundler(unpackerversion, inpart, None)

1786

# the source and url passed here are overwritten by the one contained in

1786

# the source and url passed here are overwritten by the one contained in

1787

# the transaction.hookargs argument. So 'bundle2' is a placeholder

1787

# the transaction.hookargs argument. So 'bundle2' is a placeholder

1788

nbchangesets = None

1788

nbchangesets = None

1789

if 'nbchanges' in inpart.params:

1789

if 'nbchanges' in inpart.params:

1790

nbchangesets = int(inpart.params.get('nbchanges'))

1790

nbchangesets = int(inpart.params.get('nbchanges'))

1791

if ('treemanifest' in inpart.params and

1791

if ('treemanifest' in inpart.params and

1792

'treemanifest' not in op.repo.requirements):

1792

'treemanifest' not in op.repo.requirements):

1793

if len(op.repo.changelog) != 0:

1793

if len(op.repo.changelog) != 0:

1794

raise error.Abort(_(

1794

raise error.Abort(_(

1795

"bundle contains tree manifests, but local repo is "

1795

"bundle contains tree manifests, but local repo is "

1796

"non-empty and does not use tree manifests"))

1796

"non-empty and does not use tree manifests"))

1797

op.repo.requirements.add('treemanifest')

1797

op.repo.requirements.add('treemanifest')

1798

op.repo._applyopenerreqs()

1798

op.repo._applyopenerreqs()

1799

op.repo._writerequirements()

1799

op.repo._writerequirements()

1800

extrakwargs = {}

1800

extrakwargs = {}

1801

targetphase = inpart.params.get('targetphase')

1801

targetphase = inpart.params.get('targetphase')

1802

if targetphase is not None:

1802

if targetphase is not None:

1803

extrakwargs[r'targetphase'] = int(targetphase)

1803

extrakwargs[r'targetphase'] = int(targetphase)

1804

ret = _processchangegroup(op, cg, tr, 'bundle2', 'bundle2',

1804

ret = _processchangegroup(op, cg, tr, 'bundle2', 'bundle2',

1805

expectedtotal=nbchangesets, **extrakwargs)

1805

expectedtotal=nbchangesets, **extrakwargs)

1806

if op.reply is not None:

1806

if op.reply is not None:

1807

# This is definitely not the final form of this

1807

# This is definitely not the final form of this

1808

# return. But one need to start somewhere.

1808

# return. But one need to start somewhere.

1809

part = op.reply.newpart('reply:changegroup', mandatory=False)

1809

part = op.reply.newpart('reply:changegroup', mandatory=False)

1810

part.addparam(

1810

part.addparam(

1811

'in-reply-to', pycompat.bytestr(inpart.id), mandatory=False)

1811

'in-reply-to', pycompat.bytestr(inpart.id), mandatory=False)

1812

part.addparam('return', '%i' % ret, mandatory=False)

1812

part.addparam('return', '%i' % ret, mandatory=False)

1813

assert not inpart.read()

1813

assert not inpart.read()

1814

1815

_remotechangegroupparams = tuple(['url', 'size', 'digests'] +

1815

_remotechangegroupparams = tuple(['url', 'size', 'digests'] +

1816

['digest:%s' % k for k in util.DIGESTS.keys()])

1816

['digest:%s' % k for k in util.DIGESTS.keys()])

1817

@parthandler('remote-changegroup', _remotechangegroupparams)

1817

@parthandler('remote-changegroup', _remotechangegroupparams)

1818

def handleremotechangegroup(op, inpart):

1818

def handleremotechangegroup(op, inpart):

1819

"""apply a bundle10 on the repo, given an url and validation information

1819

"""apply a bundle10 on the repo, given an url and validation information

1820

1821

All the information about the remote bundle to import are given as

1821

All the information about the remote bundle to import are given as

1822

parameters. The parameters include:

1822

parameters. The parameters include:

1823

- url: the url to the bundle10.

1823

- url: the url to the bundle10.

1824

- size: the bundle10 file size. It is used to validate what was

1824

- size: the bundle10 file size. It is used to validate what was

1825

retrieved by the client matches the server knowledge about the bundle.

1825

retrieved by the client matches the server knowledge about the bundle.

1826

- digests: a space separated list of the digest types provided as

1826

- digests: a space separated list of the digest types provided as

1827

parameters.

1827

parameters.

1828

- digest:<digest-type>: the hexadecimal representation of the digest with

1828

- digest:<digest-type>: the hexadecimal representation of the digest with

1829

that name. Like the size, it is used to validate what was retrieved by

1829

that name. Like the size, it is used to validate what was retrieved by

1830

the client matches what the server knows about the bundle.

1830

the client matches what the server knows about the bundle.

1831

1832

When multiple digest types are given, all of them are checked.

1832

When multiple digest types are given, all of them are checked.

1833

"""

1833

"""

1834

try:

1834

try:

1835

raw_url = inpart.params['url']

1835

raw_url = inpart.params['url']

1836

except KeyError:

1836

except KeyError:

1837

raise error.Abort(_('remote-changegroup: missing "%s" param') % 'url')

1837

raise error.Abort(_('remote-changegroup: missing "%s" param') % 'url')

1838

parsed_url = util.url(raw_url)

1838

parsed_url = util.url(raw_url)

1839

if parsed_url.scheme not in capabilities['remote-changegroup']:

1839

if parsed_url.scheme not in capabilities['remote-changegroup']:

1840

raise error.Abort(_('remote-changegroup does not support %s urls') %

1840

raise error.Abort(_('remote-changegroup does not support %s urls') %

1841

parsed_url.scheme)

1841

parsed_url.scheme)

1842

1843

try:

1843

try:

1844

size = int(inpart.params['size'])

1844

size = int(inpart.params['size'])

1845

except ValueError:

1845

except ValueError:

1846

raise error.Abort(_('remote-changegroup: invalid value for param "%s"')

1846

raise error.Abort(_('remote-changegroup: invalid value for param "%s"')

1847

% 'size')

1847

% 'size')

1848

except KeyError:

1848

except KeyError:

1849

raise error.Abort(_('remote-changegroup: missing "%s" param') % 'size')

1849

raise error.Abort(_('remote-changegroup: missing "%s" param') % 'size')

1850

1851

digests = {}

1851

digests = {}

1852

for typ in inpart.params.get('digests', '').split():

1852

for typ in inpart.params.get('digests', '').split():

1853

param = 'digest:%s' % typ

1853

param = 'digest:%s' % typ

1854

try:

1854

try:

1855

value = inpart.params[param]

1855

value = inpart.params[param]

1856

except KeyError:

1856

except KeyError:

1857

raise error.Abort(_('remote-changegroup: missing "%s" param') %

1857

raise error.Abort(_('remote-changegroup: missing "%s" param') %

1858

param)

1858

param)

1859

digests[typ] = value

1859

digests[typ] = value

1860

1861

real_part = util.digestchecker(url.open(op.ui, raw_url), size, digests)

1861

real_part = util.digestchecker(url.open(op.ui, raw_url), size, digests)

1862

1863

tr = op.gettransaction()

1863

tr = op.gettransaction()

1864

from . import exchange

1864

from . import exchange

1865

cg = exchange.readbundle(op.repo.ui, real_part, raw_url)

1865

cg = exchange.readbundle(op.repo.ui, real_part, raw_url)

1866

if not isinstance(cg, changegroup.cg1unpacker):

1866

if not isinstance(cg, changegroup.cg1unpacker):

1867

raise error.Abort(_('%s: not a bundle version 1.0') %

1867

raise error.Abort(_('%s: not a bundle version 1.0') %

1868

util.hidepassword(raw_url))

1868

util.hidepassword(raw_url))

1869

ret = _processchangegroup(op, cg, tr, 'bundle2', 'bundle2')

1869

ret = _processchangegroup(op, cg, tr, 'bundle2', 'bundle2')

1870

if op.reply is not None:

1870

if op.reply is not None:

1871

# This is definitely not the final form of this

1871

# This is definitely not the final form of this

1872

# return. But one need to start somewhere.

1872

# return. But one need to start somewhere.

1873

part = op.reply.newpart('reply:changegroup')

1873

part = op.reply.newpart('reply:changegroup')

1874

part.addparam(

1874

part.addparam(

1875

'in-reply-to', pycompat.bytestr(inpart.id), mandatory=False)

1875

'in-reply-to', pycompat.bytestr(inpart.id), mandatory=False)

1876

part.addparam('return', '%i' % ret, mandatory=False)

1876

part.addparam('return', '%i' % ret, mandatory=False)

1877

try:

1877

try:

1878

real_part.validate()

1878

real_part.validate()

1879

except error.Abort as e:

1879

except error.Abort as e:

1880

raise error.Abort(_('bundle at %s is corrupted:\n%s') %

1880

raise error.Abort(_('bundle at %s is corrupted:\n%s') %

1881

(util.hidepassword(raw_url), ~~str~~(e)))

1881

(util.hidepassword(raw_url), bytes(e)))

1882

assert not inpart.read()

1882

assert not inpart.read()

1883

1884

@parthandler('reply:changegroup', ('return', 'in-reply-to'))

1884

@parthandler('reply:changegroup', ('return', 'in-reply-to'))

1885

def handlereplychangegroup(op, inpart):

1885

def handlereplychangegroup(op, inpart):

1886

ret = int(inpart.params['return'])

1886

ret = int(inpart.params['return'])

1887

replyto = int(inpart.params['in-reply-to'])

1887

replyto = int(inpart.params['in-reply-to'])

1888

op.records.add('changegroup', {'return': ret}, replyto)

1888

op.records.add('changegroup', {'return': ret}, replyto)

1889

1890

@parthandler('check:bookmarks')

1890

@parthandler('check:bookmarks')

1891

def handlecheckbookmarks(op, inpart):

1891

def handlecheckbookmarks(op, inpart):

1892

"""check location of bookmarks

1892

"""check location of bookmarks

1893

1894

This part is to be used to detect push race regarding bookmark, it

1894

This part is to be used to detect push race regarding bookmark, it

1895

contains binary encoded (bookmark, node) tuple. If the local state does

1895

contains binary encoded (bookmark, node) tuple. If the local state does

1896

not marks the one in the part, a PushRaced exception is raised

1896

not marks the one in the part, a PushRaced exception is raised

1897

"""

1897

"""

1898

bookdata = bookmarks.binarydecode(inpart)

1898

bookdata = bookmarks.binarydecode(inpart)

1899

1900

msgstandard = ('repository changed while pushing - please try again '

1900

msgstandard = ('repository changed while pushing - please try again '

1901

'(bookmark "%s" move from %s to %s)')

1901

'(bookmark "%s" move from %s to %s)')

1902

msgmissing = ('repository changed while pushing - please try again '

1902

msgmissing = ('repository changed while pushing - please try again '

1903

'(bookmark "%s" is missing, expected %s)')

1903

'(bookmark "%s" is missing, expected %s)')

1904

msgexist = ('repository changed while pushing - please try again '

1904

msgexist = ('repository changed while pushing - please try again '

1905

'(bookmark "%s" set on %s, expected missing)')

1905

'(bookmark "%s" set on %s, expected missing)')

1906

for book, node in bookdata:

1906

for book, node in bookdata:

1907

currentnode = op.repo._bookmarks.get(book)

1907

currentnode = op.repo._bookmarks.get(book)

1908

if currentnode != node:

1908

if currentnode != node:

1909

if node is None:

1909

if node is None:

1910

finalmsg = msgexist % (book, nodemod.short(currentnode))

1910

finalmsg = msgexist % (book, nodemod.short(currentnode))

1911

elif currentnode is None:

1911

elif currentnode is None:

1912

finalmsg = msgmissing % (book, nodemod.short(node))

1912

finalmsg = msgmissing % (book, nodemod.short(node))

1913

else:

1913

else:

1914

finalmsg = msgstandard % (book, nodemod.short(node),

1914

finalmsg = msgstandard % (book, nodemod.short(node),

1915

nodemod.short(currentnode))

1915

nodemod.short(currentnode))

1916

raise error.PushRaced(finalmsg)

1916

raise error.PushRaced(finalmsg)

1917

1918

@parthandler('check:heads')

1918

@parthandler('check:heads')

1919

def handlecheckheads(op, inpart):

1919

def handlecheckheads(op, inpart):

1920

"""check that head of the repo did not change

1920

"""check that head of the repo did not change

1921

1922

This is used to detect a push race when using unbundle.

1922

This is used to detect a push race when using unbundle.

1923

This replaces the "heads" argument of unbundle."""

1923

This replaces the "heads" argument of unbundle."""

1924

h = inpart.read(20)

1924

h = inpart.read(20)

1925

heads = []

1925

heads = []

1926

while len(h) == 20:

1926

while len(h) == 20:

1927

heads.append(h)

1927

heads.append(h)

1928

h = inpart.read(20)

1928

h = inpart.read(20)

1929

assert not h

1929

assert not h

1930

# Trigger a transaction so that we are guaranteed to have the lock now.

1930

# Trigger a transaction so that we are guaranteed to have the lock now.

1931

if op.ui.configbool('experimental', 'bundle2lazylocking'):

1931

if op.ui.configbool('experimental', 'bundle2lazylocking'):

1932

op.gettransaction()

1932

op.gettransaction()

1933

if sorted(heads) != sorted(op.repo.heads()):

1933

if sorted(heads) != sorted(op.repo.heads()):

1934

raise error.PushRaced('repository changed while pushing - '

1934

raise error.PushRaced('repository changed while pushing - '

1935

'please try again')

1935

'please try again')

1936

1937

@parthandler('check:updated-heads')

1937

@parthandler('check:updated-heads')

1938

def handlecheckupdatedheads(op, inpart):

1938

def handlecheckupdatedheads(op, inpart):

1939

"""check for race on the heads touched by a push

1939

"""check for race on the heads touched by a push

1940

1941

This is similar to 'check:heads' but focus on the heads actually updated

1941

This is similar to 'check:heads' but focus on the heads actually updated

1942

during the push. If other activities happen on unrelated heads, it is

1942

during the push. If other activities happen on unrelated heads, it is

1943

ignored.

1943

ignored.

1944

1945

This allow server with high traffic to avoid push contention as long as

1945

This allow server with high traffic to avoid push contention as long as

1946

unrelated parts of the graph are involved."""

1946

unrelated parts of the graph are involved."""

1947

h = inpart.read(20)

1947

h = inpart.read(20)

1948

heads = []

1948

heads = []

1949

while len(h) == 20:

1949

while len(h) == 20:

1950

heads.append(h)

1950

heads.append(h)

1951

h = inpart.read(20)

1951

h = inpart.read(20)

1952

assert not h

1952

assert not h

1953

# trigger a transaction so that we are guaranteed to have the lock now.

1953

# trigger a transaction so that we are guaranteed to have the lock now.

1954

if op.ui.configbool('experimental', 'bundle2lazylocking'):

1954

if op.ui.configbool('experimental', 'bundle2lazylocking'):

1955

op.gettransaction()

1955

op.gettransaction()

1956

1957

currentheads = set()

1957

currentheads = set()

1958

for ls in op.repo.branchmap().itervalues():

1958

for ls in op.repo.branchmap().itervalues():

1959

currentheads.update(ls)

1959

currentheads.update(ls)

1960

1961

for h in heads:

1961

for h in heads:

1962

if h not in currentheads:

1962

if h not in currentheads:

1963

raise error.PushRaced('repository changed while pushing - '

1963

raise error.PushRaced('repository changed while pushing - '

1964

'please try again')

1964

'please try again')

1965

1966

@parthandler('check:phases')

1966

@parthandler('check:phases')

1967

def handlecheckphases(op, inpart):

1967

def handlecheckphases(op, inpart):

1968

"""check that phase boundaries of the repository did not change

1968

"""check that phase boundaries of the repository did not change

1969

1970

This is used to detect a push race.

1970

This is used to detect a push race.

1971

"""

1971

"""

1972

phasetonodes = phases.binarydecode(inpart)

1972

phasetonodes = phases.binarydecode(inpart)

1973

unfi = op.repo.unfiltered()

1973

unfi = op.repo.unfiltered()

1974

cl = unfi.changelog

1974

cl = unfi.changelog

1975

phasecache = unfi._phasecache

1975

phasecache = unfi._phasecache

1976

msg = ('repository changed while pushing - please try again '

1976

msg = ('repository changed while pushing - please try again '

1977

'(%s is %s expected %s)')

1977

'(%s is %s expected %s)')

1978

for expectedphase, nodes in enumerate(phasetonodes):

1978

for expectedphase, nodes in enumerate(phasetonodes):

1979

for n in nodes:

1979

for n in nodes:

1980

actualphase = phasecache.phase(unfi, cl.rev(n))

1980

actualphase = phasecache.phase(unfi, cl.rev(n))

1981

if actualphase != expectedphase:

1981

if actualphase != expectedphase:

1982

finalmsg = msg % (nodemod.short(n),

1982

finalmsg = msg % (nodemod.short(n),

1983

phases.phasenames[actualphase],

1983

phases.phasenames[actualphase],

1984

phases.phasenames[expectedphase])

1984

phases.phasenames[expectedphase])

1985

raise error.PushRaced(finalmsg)

1985

raise error.PushRaced(finalmsg)

1986

1987

@parthandler('output')

1987

@parthandler('output')

1988

def handleoutput(op, inpart):

1988

def handleoutput(op, inpart):

1989

"""forward output captured on the server to the client"""

1989

"""forward output captured on the server to the client"""

1990

for line in inpart.read().splitlines():

1990

for line in inpart.read().splitlines():

1991

op.ui.status(_('remote: %s\n') % line)

1991

op.ui.status(_('remote: %s\n') % line)

1992

1993

@parthandler('replycaps')

1993

@parthandler('replycaps')

1994

def handlereplycaps(op, inpart):

1994

def handlereplycaps(op, inpart):

1995

"""Notify that a reply bundle should be created

1995

"""Notify that a reply bundle should be created

1996

1997

The payload contains the capabilities information for the reply"""

1997

The payload contains the capabilities information for the reply"""

1998

caps = decodecaps(inpart.read())

1998

caps = decodecaps(inpart.read())

1999

if op.reply is None:

1999

if op.reply is None:

2000

op.reply = bundle20(op.ui, caps)

2000

op.reply = bundle20(op.ui, caps)

2001

2002

class AbortFromPart(error.Abort):

2002

class AbortFromPart(error.Abort):

2003

"""Sub-class of Abort that denotes an error from a bundle2 part."""

2003

"""Sub-class of Abort that denotes an error from a bundle2 part."""

2004

2005

@parthandler('error:abort', ('message', 'hint'))

2005

@parthandler('error:abort', ('message', 'hint'))

2006

def handleerrorabort(op, inpart):

2006

def handleerrorabort(op, inpart):

2007

"""Used to transmit abort error over the wire"""

2007

"""Used to transmit abort error over the wire"""

2008

raise AbortFromPart(inpart.params['message'],

2008

raise AbortFromPart(inpart.params['message'],

2009

hint=inpart.params.get('hint'))

2009

hint=inpart.params.get('hint'))

2010

2011

@parthandler('error:pushkey', ('namespace', 'key', 'new', 'old', 'ret',

2011

@parthandler('error:pushkey', ('namespace', 'key', 'new', 'old', 'ret',

2012

'in-reply-to'))

2012

'in-reply-to'))

2013

def handleerrorpushkey(op, inpart):

2013

def handleerrorpushkey(op, inpart):

2014

"""Used to transmit failure of a mandatory pushkey over the wire"""

2014

"""Used to transmit failure of a mandatory pushkey over the wire"""

2015

kwargs = {}

2015

kwargs = {}

2016

for name in ('namespace', 'key', 'new', 'old', 'ret'):

2016

for name in ('namespace', 'key', 'new', 'old', 'ret'):

2017

value = inpart.params.get(name)

2017

value = inpart.params.get(name)

2018

if value is not None:

2018

if value is not None:

2019

kwargs[name] = value

2019

kwargs[name] = value

2020

raise error.PushkeyFailed(inpart.params['in-reply-to'],

2020

raise error.PushkeyFailed(inpart.params['in-reply-to'],

2021

**pycompat.strkwargs(kwargs))

2021

**pycompat.strkwargs(kwargs))

2022

2023

@parthandler('error:unsupportedcontent', ('parttype', 'params'))

2023

@parthandler('error:unsupportedcontent', ('parttype', 'params'))

2024

def handleerrorunsupportedcontent(op, inpart):

2024

def handleerrorunsupportedcontent(op, inpart):

2025

"""Used to transmit unknown content error over the wire"""

2025

"""Used to transmit unknown content error over the wire"""

2026

kwargs = {}

2026

kwargs = {}

2027

parttype = inpart.params.get('parttype')

2027

parttype = inpart.params.get('parttype')

2028

if parttype is not None:

2028

if parttype is not None:

2029

kwargs['parttype'] = parttype

2029

kwargs['parttype'] = parttype

2030

params = inpart.params.get('params')

2030

params = inpart.params.get('params')

2031

if params is not None:

2031

if params is not None:

2032

kwargs['params'] = params.split('\0')

2032

kwargs['params'] = params.split('\0')

2033

2034

raise error.BundleUnknownFeatureError(**pycompat.strkwargs(kwargs))

2034

raise error.BundleUnknownFeatureError(**pycompat.strkwargs(kwargs))

2035

2036

@parthandler('error:pushraced', ('message',))

2036

@parthandler('error:pushraced', ('message',))

2037

def handleerrorpushraced(op, inpart):

2037

def handleerrorpushraced(op, inpart):

2038

"""Used to transmit push race error over the wire"""

2038

"""Used to transmit push race error over the wire"""

2039

raise error.ResponseError(_('push failed:'), inpart.params['message'])

2039

raise error.ResponseError(_('push failed:'), inpart.params['message'])

2040

2041

@parthandler('listkeys', ('namespace',))

2041

@parthandler('listkeys', ('namespace',))

2042

def handlelistkeys(op, inpart):

2042

def handlelistkeys(op, inpart):

2043

"""retrieve pushkey namespace content stored in a bundle2"""

2043

"""retrieve pushkey namespace content stored in a bundle2"""

2044

namespace = inpart.params['namespace']

2044

namespace = inpart.params['namespace']

2045

r = pushkey.decodekeys(inpart.read())

2045

r = pushkey.decodekeys(inpart.read())

2046

op.records.add('listkeys', (namespace, r))

2046

op.records.add('listkeys', (namespace, r))

2047

2048

@parthandler('pushkey', ('namespace', 'key', 'old', 'new'))

2048

@parthandler('pushkey', ('namespace', 'key', 'old', 'new'))

2049

def handlepushkey(op, inpart):

2049

def handlepushkey(op, inpart):

2050

"""process a pushkey request"""

2050

"""process a pushkey request"""

2051

dec = pushkey.decode

2051

dec = pushkey.decode

2052

namespace = dec(inpart.params['namespace'])

2052

namespace = dec(inpart.params['namespace'])

2053

key = dec(inpart.params['key'])

2053

key = dec(inpart.params['key'])

2054

old = dec(inpart.params['old'])

2054

old = dec(inpart.params['old'])

2055

new = dec(inpart.params['new'])

2055

new = dec(inpart.params['new'])

2056

# Grab the transaction to ensure that we have the lock before performing the

2056

# Grab the transaction to ensure that we have the lock before performing the

2057

# pushkey.

2057

# pushkey.

2058

if op.ui.configbool('experimental', 'bundle2lazylocking'):

2058

if op.ui.configbool('experimental', 'bundle2lazylocking'):

2059

op.gettransaction()

2059

op.gettransaction()

2060

ret = op.repo.pushkey(namespace, key, old, new)

2060

ret = op.repo.pushkey(namespace, key, old, new)

2061

record = {'namespace': namespace,

2061

record = {'namespace': namespace,

2062

'key': key,

2062

'key': key,

2063

'old': old,

2063

'old': old,

2064

'new': new}

2064

'new': new}

2065

op.records.add('pushkey', record)

2065

op.records.add('pushkey', record)

2066

if op.reply is not None:

2066

if op.reply is not None:

2067

rpart = op.reply.newpart('reply:pushkey')

2067

rpart = op.reply.newpart('reply:pushkey')

2068

rpart.addparam(

2068

rpart.addparam(

2069

'in-reply-to', pycompat.bytestr(inpart.id), mandatory=False)

2069

'in-reply-to', pycompat.bytestr(inpart.id), mandatory=False)

2070

rpart.addparam('return', '%i' % ret, mandatory=False)

2070

rpart.addparam('return', '%i' % ret, mandatory=False)

2071

if inpart.mandatory and not ret:

2071

if inpart.mandatory and not ret:

2072

kwargs = {}

2072

kwargs = {}

2073

for key in ('namespace', 'key', 'new', 'old', 'ret'):

2073

for key in ('namespace', 'key', 'new', 'old', 'ret'):

2074

if key in inpart.params:

2074

if key in inpart.params:

2075

kwargs[key] = inpart.params[key]

2075

kwargs[key] = inpart.params[key]

2076

raise error.PushkeyFailed(partid='%d' % inpart.id,

2076

raise error.PushkeyFailed(partid='%d' % inpart.id,

2077

**pycompat.strkwargs(kwargs))

2077

**pycompat.strkwargs(kwargs))

2078

2079

@parthandler('bookmarks')

2079

@parthandler('bookmarks')

2080

def handlebookmark(op, inpart):

2080

def handlebookmark(op, inpart):

2081

"""transmit bookmark information

2081

"""transmit bookmark information

2082

2083

The part contains binary encoded bookmark information.

2083

The part contains binary encoded bookmark information.

2084

2085

The exact behavior of this part can be controlled by the 'bookmarks' mode

2085

The exact behavior of this part can be controlled by the 'bookmarks' mode

2086

on the bundle operation.

2086

on the bundle operation.

2087

2088

When mode is 'apply' (the default) the bookmark information is applied as

2088

When mode is 'apply' (the default) the bookmark information is applied as

2089

is to the unbundling repository. Make sure a 'check:bookmarks' part is

2089

is to the unbundling repository. Make sure a 'check:bookmarks' part is

2090

issued earlier to check for push races in such update. This behavior is

2090

issued earlier to check for push races in such update. This behavior is

2091

suitable for pushing.

2091

suitable for pushing.

2092

2093

When mode is 'records', the information is recorded into the 'bookmarks'

2093

When mode is 'records', the information is recorded into the 'bookmarks'

2094

records of the bundle operation. This behavior is suitable for pulling.

2094

records of the bundle operation. This behavior is suitable for pulling.

2095

"""

2095

"""

2096

changes = bookmarks.binarydecode(inpart)

2096

changes = bookmarks.binarydecode(inpart)

2097

2098

pushkeycompat = op.repo.ui.configbool('server', 'bookmarks-pushkey-compat')

2098

pushkeycompat = op.repo.ui.configbool('server', 'bookmarks-pushkey-compat')

2099

bookmarksmode = op.modes.get('bookmarks', 'apply')

2099

bookmarksmode = op.modes.get('bookmarks', 'apply')

2100

2101

if bookmarksmode == 'apply':

2101

if bookmarksmode == 'apply':

2102

tr = op.gettransaction()

2102

tr = op.gettransaction()

2103

bookstore = op.repo._bookmarks

2103

bookstore = op.repo._bookmarks

2104

if pushkeycompat:

2104

if pushkeycompat:

2105

allhooks = []

2105

allhooks = []

2106

for book, node in changes:

2106

for book, node in changes:

2107

hookargs = tr.hookargs.copy()

2107

hookargs = tr.hookargs.copy()

2108

hookargs['pushkeycompat'] = '1'

2108

hookargs['pushkeycompat'] = '1'

2109

hookargs['namespace'] = 'bookmarks'

2109

hookargs['namespace'] = 'bookmarks'

2110

hookargs['key'] = book

2110

hookargs['key'] = book

2111

hookargs['old'] = nodemod.hex(bookstore.get(book, ''))

2111

hookargs['old'] = nodemod.hex(bookstore.get(book, ''))

2112

hookargs['new'] = nodemod.hex(node if node is not None else '')

2112

hookargs['new'] = nodemod.hex(node if node is not None else '')

2113

allhooks.append(hookargs)

2113

allhooks.append(hookargs)

2114

2115

for hookargs in allhooks:

2115

for hookargs in allhooks:

2116

op.repo.hook('prepushkey', throw=True,

2116

op.repo.hook('prepushkey', throw=True,

2117

**pycompat.strkwargs(hookargs))

2117

**pycompat.strkwargs(hookargs))

2118

2119

bookstore.applychanges(op.repo, op.gettransaction(), changes)

2119

bookstore.applychanges(op.repo, op.gettransaction(), changes)

2120

2121

if pushkeycompat:

2121

if pushkeycompat:

2122

def runhook():

2122

def runhook():

2123

for hookargs in allhooks:

2123

for hookargs in allhooks:

2124

op.repo.hook('pushkey', **pycompat.strkwargs(hookargs))

2124

op.repo.hook('pushkey', **pycompat.strkwargs(hookargs))

2125

op.repo._afterlock(runhook)

2125

op.repo._afterlock(runhook)

2126

2127

elif bookmarksmode == 'records':

2127

elif bookmarksmode == 'records':

2128

for book, node in changes:

2128

for book, node in changes:

2129

record = {'bookmark': book, 'node': node}

2129

record = {'bookmark': book, 'node': node}

2130

op.records.add('bookmarks', record)

2130

op.records.add('bookmarks', record)

2131

else:

2131

else:

2132

raise error.ProgrammingError('unkown bookmark mode: %s' % bookmarksmode)

2132

raise error.ProgrammingError('unkown bookmark mode: %s' % bookmarksmode)

2133

2134

@parthandler('phase-heads')

2134

@parthandler('phase-heads')

2135

def handlephases(op, inpart):

2135

def handlephases(op, inpart):

2136

"""apply phases from bundle part to repo"""

2136

"""apply phases from bundle part to repo"""

2137

headsbyphase = phases.binarydecode(inpart)

2137

headsbyphase = phases.binarydecode(inpart)

2138

phases.updatephases(op.repo.unfiltered(), op.gettransaction, headsbyphase)

2138

phases.updatephases(op.repo.unfiltered(), op.gettransaction, headsbyphase)

2139

2140

@parthandler('reply:pushkey', ('return', 'in-reply-to'))

2140

@parthandler('reply:pushkey', ('return', 'in-reply-to'))

2141

def handlepushkeyreply(op, inpart):

2141

def handlepushkeyreply(op, inpart):

2142

"""retrieve the result of a pushkey request"""

2142

"""retrieve the result of a pushkey request"""

2143

ret = int(inpart.params['return'])

2143

ret = int(inpart.params['return'])

2144

partid = int(inpart.params['in-reply-to'])

2144

partid = int(inpart.params['in-reply-to'])

2145

op.records.add('pushkey', {'return': ret}, partid)

2145

op.records.add('pushkey', {'return': ret}, partid)

2146

2147

@parthandler('obsmarkers')

2147

@parthandler('obsmarkers')

2148

def handleobsmarker(op, inpart):

2148

def handleobsmarker(op, inpart):

2149

"""add a stream of obsmarkers to the repo"""

2149

"""add a stream of obsmarkers to the repo"""

2150

tr = op.gettransaction()

2150

tr = op.gettransaction()

2151

markerdata = inpart.read()

2151

markerdata = inpart.read()

2152

if op.ui.config('experimental', 'obsmarkers-exchange-debug'):

2152

if op.ui.config('experimental', 'obsmarkers-exchange-debug'):

2153

op.ui.write(('obsmarker-exchange: %i bytes received\n')

2153

op.ui.write(('obsmarker-exchange: %i bytes received\n')

2154

% len(markerdata))

2154

% len(markerdata))

2155

# The mergemarkers call will crash if marker creation is not enabled.

2155

# The mergemarkers call will crash if marker creation is not enabled.

2156

# we want to avoid this if the part is advisory.

2156

# we want to avoid this if the part is advisory.

2157

if not inpart.mandatory and op.repo.obsstore.readonly:

2157

if not inpart.mandatory and op.repo.obsstore.readonly:

2158

op.repo.ui.debug('ignoring obsolescence markers, feature not enabled\n')

2158

op.repo.ui.debug('ignoring obsolescence markers, feature not enabled\n')

2159

return

2159

return

2160

new = op.repo.obsstore.mergemarkers(tr, markerdata)

2160

new = op.repo.obsstore.mergemarkers(tr, markerdata)

2161

op.repo.invalidatevolatilesets()

2161

op.repo.invalidatevolatilesets()

2162

if new:

2162

if new:

2163

op.repo.ui.status(_('%i new obsolescence markers\n') % new)

2163

op.repo.ui.status(_('%i new obsolescence markers\n') % new)

2164

op.records.add('obsmarkers', {'new': new})

2164

op.records.add('obsmarkers', {'new': new})

2165

if op.reply is not None:

2165

if op.reply is not None:

2166

rpart = op.reply.newpart('reply:obsmarkers')

2166

rpart = op.reply.newpart('reply:obsmarkers')

2167

rpart.addparam(

2167

rpart.addparam(

2168

'in-reply-to', pycompat.bytestr(inpart.id), mandatory=False)

2168

'in-reply-to', pycompat.bytestr(inpart.id), mandatory=False)

2169

rpart.addparam('new', '%i' % new, mandatory=False)

2169

rpart.addparam('new', '%i' % new, mandatory=False)

2170

2171

2172

@parthandler('reply:obsmarkers', ('new', 'in-reply-to'))

2172

@parthandler('reply:obsmarkers', ('new', 'in-reply-to'))

2173

def handleobsmarkerreply(op, inpart):

2173

def handleobsmarkerreply(op, inpart):

2174

"""retrieve the result of a pushkey request"""

2174

"""retrieve the result of a pushkey request"""

2175

ret = int(inpart.params['new'])

2175

ret = int(inpart.params['new'])

2176

partid = int(inpart.params['in-reply-to'])

2176

partid = int(inpart.params['in-reply-to'])

2177

op.records.add('obsmarkers', {'new': ret}, partid)

2177

op.records.add('obsmarkers', {'new': ret}, partid)

2178

2179

@parthandler('hgtagsfnodes')

2179

@parthandler('hgtagsfnodes')

2180

def handlehgtagsfnodes(op, inpart):

2180

def handlehgtagsfnodes(op, inpart):

2181

"""Applies .hgtags fnodes cache entries to the local repo.

2181

"""Applies .hgtags fnodes cache entries to the local repo.

2182

2183

Payload is pairs of 20 byte changeset nodes and filenodes.

2183

Payload is pairs of 20 byte changeset nodes and filenodes.

2184

"""

2184

"""

2185

# Grab the transaction so we ensure that we have the lock at this point.

2185

# Grab the transaction so we ensure that we have the lock at this point.

2186

if op.ui.configbool('experimental', 'bundle2lazylocking'):

2186

if op.ui.configbool('experimental', 'bundle2lazylocking'):

2187

op.gettransaction()

2187

op.gettransaction()

2188

cache = tags.hgtagsfnodescache(op.repo.unfiltered())

2188

cache = tags.hgtagsfnodescache(op.repo.unfiltered())

2189

2190

count = 0

2190

count = 0

2191

while True:

2191

while True:

2192

node = inpart.read(20)

2192

node = inpart.read(20)

2193

fnode = inpart.read(20)

2193

fnode = inpart.read(20)

2194

if len(node) < 20 or len(fnode) < 20:

2194

if len(node) < 20 or len(fnode) < 20:

2195

op.ui.debug('ignoring incomplete received .hgtags fnodes data\n')

2195

op.ui.debug('ignoring incomplete received .hgtags fnodes data\n')

2196

break

2196

break

2197

cache.setfnode(node, fnode)

2197

cache.setfnode(node, fnode)

2198

count += 1

2198

count += 1

2199

2200

cache.write()

2200

cache.write()

2201

op.ui.debug('applied %i hgtags fnodes cache entries\n' % count)

2201

op.ui.debug('applied %i hgtags fnodes cache entries\n' % count)

2202

2203

rbcstruct = struct.Struct('>III')

2203

rbcstruct = struct.Struct('>III')

2204

2205

@parthandler('cache:rev-branch-cache')

2205

@parthandler('cache:rev-branch-cache')

2206

def handlerbc(op, inpart):

2206

def handlerbc(op, inpart):

2207

"""receive a rev-branch-cache payload and update the local cache

2207

"""receive a rev-branch-cache payload and update the local cache

2208

2209

The payload is a series of data related to each branch

2209

The payload is a series of data related to each branch

2210

2211

1) branch name length

2211

1) branch name length

2212

2) number of open heads

2212

2) number of open heads

2213

3) number of closed heads

2213

3) number of closed heads

2214

4) open heads nodes

2214

4) open heads nodes

2215

5) closed heads nodes

2215

5) closed heads nodes

2216

"""

2216

"""

2217

total = 0

2217

total = 0

2218

rawheader = inpart.read(rbcstruct.size)

2218

rawheader = inpart.read(rbcstruct.size)

2219

cache = op.repo.revbranchcache()

2219

cache = op.repo.revbranchcache()

2220

cl = op.repo.unfiltered().changelog

2220

cl = op.repo.unfiltered().changelog

2221

while rawheader:

2221

while rawheader:

2222

header = rbcstruct.unpack(rawheader)

2222

header = rbcstruct.unpack(rawheader)

2223

total += header[1] + header[2]

2223

total += header[1] + header[2]

2224

utf8branch = inpart.read(header[0])

2224

utf8branch = inpart.read(header[0])

2225

branch = encoding.tolocal(utf8branch)

2225

branch = encoding.tolocal(utf8branch)

2226

for x in xrange(header[1]):

2226

for x in xrange(header[1]):

2227

node = inpart.read(20)

2227

node = inpart.read(20)

2228

rev = cl.rev(node)

2228

rev = cl.rev(node)

2229

cache.setdata(branch, rev, node, False)

2229

cache.setdata(branch, rev, node, False)

2230

for x in xrange(header[2]):

2230

for x in xrange(header[2]):

2231

node = inpart.read(20)

2231

node = inpart.read(20)

2232

rev = cl.rev(node)

2232

rev = cl.rev(node)

2233

cache.setdata(branch, rev, node, True)

2233

cache.setdata(branch, rev, node, True)

2234

rawheader = inpart.read(rbcstruct.size)

2234

rawheader = inpart.read(rbcstruct.size)

2235

cache.write()

2235

cache.write()

2236

2237

@parthandler('pushvars')

2237

@parthandler('pushvars')

2238

def bundle2getvars(op, part):

2238

def bundle2getvars(op, part):

2239

'''unbundle a bundle2 containing shellvars on the server'''

2239

'''unbundle a bundle2 containing shellvars on the server'''

2240

# An option to disable unbundling on server-side for security reasons

2240

# An option to disable unbundling on server-side for security reasons

2241

if op.ui.configbool('push', 'pushvars.server'):

2241

if op.ui.configbool('push', 'pushvars.server'):

2242

hookargs = {}

2242

hookargs = {}

2243

for key, value in part.advisoryparams:

2243

for key, value in part.advisoryparams:

2244

key = key.upper()

2244

key = key.upper()

2245

# We want pushed variables to have USERVAR_ prepended so we know

2245

# We want pushed variables to have USERVAR_ prepended so we know

2246

# they came from the --pushvar flag.

2246

# they came from the --pushvar flag.

2247

key = "USERVAR_" + key

2247

key = "USERVAR_" + key

2248

hookargs[key] = value

2248

hookargs[key] = value

2249

op.addhookargs(hookargs)

2249

op.addhookargs(hookargs)

2250

2251

@parthandler('stream2', ('requirements', 'filecount', 'bytecount'))

2251

@parthandler('stream2', ('requirements', 'filecount', 'bytecount'))

2252

def handlestreamv2bundle(op, part):

2252

def handlestreamv2bundle(op, part):

2253

2254

requirements = urlreq.unquote(part.params['requirements']).split(',')

2254

requirements = urlreq.unquote(part.params['requirements']).split(',')

2255

filecount = int(part.params['filecount'])

2255

filecount = int(part.params['filecount'])

2256

bytecount = int(part.params['bytecount'])

2256

bytecount = int(part.params['bytecount'])

2257

2258

repo = op.repo

2258

repo = op.repo

2259

if len(repo):

2259

if len(repo):

2260

msg = _('cannot apply stream clone to non empty repository')

2260

msg = _('cannot apply stream clone to non empty repository')

2261

raise error.Abort(msg)

2261

raise error.Abort(msg)

2262

2263

repo.ui.debug('applying stream bundle\n')

2263

repo.ui.debug('applying stream bundle\n')

2264

streamclone.applybundlev2(repo, part, filecount, bytecount,

2264

streamclone.applybundlev2(repo, part, filecount, bytecount,

2265

requirements)

2265

requirements)

	Site-wide shortcuts
/	Use quick search box
g h	Goto home page
g g	Goto my private gists page
g G	Goto my public gists page
g 0-9	Goto bookmarked items from 0-9
n r	New repository page
n g	New gist page

	Repositories
g s	Goto summary page
g c	Goto changelog page
g f	Goto files page
g F	Goto files page with file search activated
g p	Goto pull requests page
g o	Goto repository settings
g O	Goto repository access permissions settings
t s	Toggle sidebar on some pages

             test-abort-checkin.t
             test-add.t
             test-addremove-similar.t
             test-addremove.t
             test-alias.t
             test-amend-subrepo.t
             test-amend.t
             test-ancestor.py
             test-annotate.py
             test-annotate.t
             test-archive-symlinks.t
             test-atomictempfile.py
             test-audit-path.t
             test-audit-subrepo.t
             test-automv.t
             test-backout.t
             test-backwards-remove.t
             test-bad-pull.t
             test-basic.t
             test-bheads.t
             test-bisect.t
             test-bisect2.t
             test-bisect3.t
             test-blackbox.t
             test-bookmarks-current.t
             test-bookmarks-merge.t
             test-bookmarks-pushpull.t
             test-bookmarks-rebase.t
             test-bookmarks-strip.t
             test-bookmarks.t
             test-branch-change.t
             test-branch-option.t
             test-branch-tag-confict.t
             test-branches.t
             test-bundle-phases.t
             test-bundle-r.t
             test-bundle-type.t
             test-bundle-vs-outgoing.t
             test-bundle.t
             test-bundle2-exchange.t
             test-bundle2-format.t
             test-bundle2-multiple-changegroups.t
             test-bundle2-pushback.t
+            test-bundle2-remote-changegroup.t
             test-cappedreader.py
             test-casecollision.t
             test-cat.t
             test-cbor.py
             test-censor.t
             test-changelog-exec.t
             test-check-commit.t
             test-check-execute.t
             test-check-interfaces.py
             test-check-module-imports.t
             test-check-pyflakes.t
             test-check-pylint.t
             test-check-shbang.t
             test-children.t
             test-clone-cgi.t
             test-clone-pull-corruption.t
             test-clone-r.t
             test-clone-update-order.t
             test-clonebundles.t
             test-commit-amend.t
             test-commit-interactive.t
             test-commit-multiple.t
             test-commit-unresolved.t
             test-commit.t
             test-committer.t
             test-completion.t
             test-config-env.py
             test-config.t
             test-conflict.t
             test-confused-revert.t
             test-context.py
             test-contrib-check-code.t
             test-contrib-check-commit.t
             test-convert-authormap.t
             test-convert-clonebranches.t
             test-convert-cvs-branch.t
             test-convert-cvs-detectmerge.t
             test-convert-cvs-synthetic.t
             test-convert-cvs.t
             test-convert-cvsnt-mergepoints.t
             test-convert-datesort.t
             test-convert-filemap.t
             test-convert-hg-sink.t
             test-convert-hg-source.t
             test-convert-hg-startrev.t
             test-convert-splicemap.t
             test-convert-tagsbranch-topology.t
             test-copy-move-merge.t
             test-copy.t
             test-copytrace-heuristics.t
             test-debugbuilddag.t
             test-debugbundle.t
             test-debugextensions.t
             test-debugindexdot.t
             test-debugrename.t
             test-default-push.t
             test-diff-antipatience.t
             test-diff-binary-file.t
             test-diff-change.t
             test-diff-copy-depth.t
             test-diff-hashes.t
             test-diff-ignore-whitespace.t
             test-diff-indent-heuristic.t
             test-diff-issue2761.t
             test-diff-newlines.t
             test-diff-reverse.t
             test-diff-subdir.t
             test-diff-unified.t
             test-diff-upgrade.t
             test-diffdir.t
             test-diffstat.t
             test-directaccess.t
             test-dirstate-backup.t
             test-dirstate-nonnormalset.t
             test-dirstate.t
             test-dispatch.py
             test-doctest.py
             test-double-merge.t
             test-drawdag.t
             test-duplicateoptions.py
             test-editor-filename.t
             test-empty-dir.t
             test-empty-file.t
             test-empty-group.t
             test-empty.t
             test-encode.t
             test-encoding-func.py
             test-encoding.t
             test-eol-add.t
             test-eol-clone.t
             test-eol-hook.t
             test-eol-patch.t
             test-eol-tag.t
             test-eol-update.t
             test-eol.t
             test-eolfilename.t
             test-excessive-merge.t
             test-exchange-obsmarkers-case-A1.t
             test-exchange-obsmarkers-case-A2.t
             test-exchange-obsmarkers-case-A3.t
             test-exchange-obsmarkers-case-A4.t
             test-exchange-obsmarkers-case-A5.t
             test-exchange-obsmarkers-case-A6.t
             test-exchange-obsmarkers-case-A7.t
             test-exchange-obsmarkers-case-B1.t
             test-exchange-obsmarkers-case-B2.t
             test-exchange-obsmarkers-case-B3.t
             test-exchange-obsmarkers-case-B4.t
             test-exchange-obsmarkers-case-B5.t
             test-exchange-obsmarkers-case-B6.t
             test-exchange-obsmarkers-case-B7.t
             test-exchange-obsmarkers-case-C1.t
             test-exchange-obsmarkers-case-C2.t
             test-exchange-obsmarkers-case-C3.t
             test-exchange-obsmarkers-case-C4.t
             test-exchange-obsmarkers-case-D1.t
             test-exchange-obsmarkers-case-D2.t
             test-exchange-obsmarkers-case-D3.t
             test-exchange-obsmarkers-case-D4.t
             test-execute-bit.t
             test-export.t
             test-extdata.t
             test-extdiff.t
             test-extensions-afterloaded.t
             test-extensions-wrapfunction.py
             test-extra-filelog-entry.t
             test-fetch.t
             test-filebranch.t
             test-filecache.py
             test-filelog.py
             test-fileset-generated.t
             test-fileset.t
             test-fix-topology.t
             test-flags.t
             test-generaldelta.t
             test-getbundle.t
             test-git-export.t
             test-glog-topological.t
             test-gpg.t
             test-graft.t
             test-hg-parseurl.py
             test-hghave.t
             test-hgignore.t
             test-hgk.t
             test-hgrc.t
             test-hgweb-bundle.t
             test-hgweb-descend-empties.t
             test-hgweb-empty.t
             test-hgweb-removed.t
             test-hgwebdir-paths.py
             test-hgwebdirsym.t
             test-histedit-arguments.t
             test-histedit-base.t
             test-histedit-bookmark-motion.t
             test-histedit-commute.t
             test-histedit-drop.t
             test-histedit-edit.t
             test-histedit-fold-non-commute.t
             test-histedit-fold.t
             test-histedit-no-backup.t
             test-histedit-no-change.t
             test-histedit-non-commute-abort.t
             test-histedit-non-commute.t
             test-histedit-obsolete.t
             test-histedit-outgoing.t
             test-histedit-templates.t
             test-http-branchmap.t
             test-http-bundle1.t
             test-http-clone-r.t
             test-http.t
             test-hybridencode.py
             test-identify.t
             test-impexp-branch.t
             test-import-bypass.t
             test-import-eol.t
             test-import-merge.t
             test-import-unknown.t
             test-import.t
             test-imports-checker.t
             test-incoming-outgoing.t
             test-inherit-mode.t
             test-init.t
             test-issue1089.t
             test-issue1102.t
             test-issue1175.t
             test-issue1306.t
             test-issue1438.t
             test-issue1502.t
             test-issue1802.t
             test-issue1877.t
             test-issue1993.t
             test-issue2137.t
             test-issue3084.t
             test-issue4074.t
             test-issue522.t
             test-issue586.t
             test-issue612.t
             test-issue619.t
             test-issue660.t
             test-issue672.t
             test-issue842.t
             test-journal-exists.t
             test-journal-share.t
             test-journal.t
             test-known.t
             test-largefiles-cache.t
             test-largefiles-misc.t
             test-largefiles-small-disk.t
             test-largefiles-update.t
             test-largefiles.t
             test-lfs-largefiles.t
             test-lfs-pointer.py
             test-linerange.py
             test-locate.t
             test-lock-badness.t
             test-log-linerange.t
             test-log.t
             test-logexchange.t
             test-lrucachedict.py
             test-mactext.t
             test-mailmap.t
             test-manifest-merging.t
             test-manifest.py
             test-manifest.t
             test-match.py
             test-mdiff.py
             test-merge-changedelete.t
             test-merge-closedheads.t
             test-merge-commit.t
             test-merge-criss-cross.t
             test-merge-default.t
             test-merge-force.t
             test-merge-halt.t
             test-merge-internal-tools-pattern.t
             test-merge-local.t
             test-merge-remove.t
             test-merge-revert.t
             test-merge-revert2.t
             test-merge-subrepos.t
             test-merge-symlinks.t
             test-merge-tools.t
             test-merge-types.t
             test-merge1.t
             test-merge10.t
             test-merge2.t
             test-merge4.t
             test-merge5.t
             test-merge6.t
             test-merge7.t
             test-merge8.t
             test-merge9.t
             test-minifileset.py
             test-minirst.py
             test-mq-git.t
             test-mq-header-date.t
             test-mq-header-from.t
             test-mq-merge.t
             test-mq-pull-from-bundle.t
             test-mq-qclone-http.t
             test-mq-qdelete.t
             test-mq-qdiff.t
             test-mq-qfold.t
             test-mq-qgoto.t
             test-mq-qimport-fail-cleanup.t
             test-mq-qnew.t
             test-mq-qpush-exact.t
             test-mq-qqueue.t
             test-mq-qrefresh-interactive.t
             test-mq-qrefresh-replace-log-message.t
             test-mq-qrefresh.t
             test-mq-qrename.t
             test-mq-qsave.t
             test-mq-safety.t
             test-mq-subrepo.t
             test-mq-symlinks.t
             test-mv-cp-st-diff.t
             test-narrow-archive.t
             test-narrow-clone-no-ellipsis.t
             test-narrow-clone-non-narrow-server.t
             test-narrow-clone-nonlinear.t
             test-narrow-clone.t
             test-narrow-commit.t
             test-narrow-copies.t
             test-narrow-debugcommands.t
             test-narrow-debugrebuilddirstate.t
             test-narrow-exchange-merges.t
             test-narrow-exchange.t
             test-narrow-expanddirstate.t
             test-narrow-merge.t
             test-narrow-patch.t
             test-narrow-patterns.t
             test-narrow-pull.t
             test-narrow-rebase.t
             test-narrow-shallow-merges.t
             test-narrow-shallow.t
             test-narrow-strip.t
             test-narrow-update.t
             test-narrow-widen.t
             test-narrow.t
             test-nested-repo.t
             test-newbranch.t
             test-nointerrupt.t
             test-obshistory.t
             test-obsmarker-template.t
             test-obsmarkers-effectflag.t
             test-obsolete-bundle-strip.t
             test-obsolete-changeset-exchange.t
             test-obsolete-checkheads.t
             test-obsolete-distributed.t
             test-obsolete-divergent.t
             test-obsolete-tag-cache.t
             test-pager.t
             test-parents.t
             test-parseindex2.py
             test-patch-offset.t
             test-patch.t
             test-pathconflicts-merge.t
             test-pathconflicts-update.t
             test-pathencode.py
             test-pending.t
             test-permissions.t
             test-phases.t
             test-pull-branch.t
             test-pull-http.t
             test-pull-permission.t
             test-pull-pull-corruption.t
             test-pull-r.t
             test-pull-update.t
             test-pull.t
             test-purge.t
             test-push-checkheads-partial-C1.t
             test-push-checkheads-partial-C2.t
             test-push-checkheads-partial-C3.t
             test-push-checkheads-partial-C4.t
             test-push-checkheads-pruned-B1.t
             test-push-checkheads-pruned-B2.t
             test-push-checkheads-pruned-B3.t
             test-push-checkheads-pruned-B4.t
             test-push-checkheads-pruned-B5.t
             test-push-checkheads-pruned-B6.t
             test-push-checkheads-pruned-B7.t
             test-push-checkheads-pruned-B8.t
             test-push-checkheads-superceed-A1.t
             test-push-checkheads-superceed-A2.t
             test-push-checkheads-superceed-A3.t
             test-push-checkheads-superceed-A4.t
             test-push-checkheads-superceed-A5.t
             test-push-checkheads-superceed-A6.t
             test-push-checkheads-superceed-A7.t
             test-push-checkheads-superceed-A8.t
             test-push-checkheads-unpushed-D1.t
             test-push-checkheads-unpushed-D2.t
             test-push-checkheads-unpushed-D3.t
             test-push-checkheads-unpushed-D4.t
             test-push-checkheads-unpushed-D5.t
             test-push-checkheads-unpushed-D6.t
             test-push-checkheads-unpushed-D7.t
             test-push-http.t
             test-push-warn.t
             test-push.t
             test-pushvars.t
             test-qrecord.t
             test-rebase-abort.t
             test-rebase-base-flag.t
             test-rebase-bookmarks.t
             test-rebase-brute-force.t
             test-rebase-cache.t
             test-rebase-check-restore.t
             test-rebase-collapse.t
             test-rebase-conflicts.t
             test-rebase-dest.t
             test-rebase-detach.t
             test-rebase-emptycommit.t
             test-rebase-inmemory.t
             test-rebase-interruptions.t
             test-rebase-issue-noparam-single-rev.t
             test-rebase-legacy.t
             test-rebase-mq-skip.t
             test-rebase-mq.t
             test-rebase-named-branches.t
             test-rebase-newancestor.t
             test-rebase-obsolete.t
             test-rebase-parameters.t
             test-rebase-partial.t
             test-rebase-pull.t
             test-rebase-rename.t
             test-rebase-scenario-global.t
             test-rebase-templates.t
             test-rebase-transaction.t
             test-rebuildstate.t
             test-record.t
             test-relink.t
             test-remove.t
             test-removeemptydirs.t
             test-rename-after-merge.t
             test-rename-dir-merge.t
             test-rename-merge1.t
             test-rename.t
             test-repair-strip.t
             test-repo-compengines.t
             test-resolve.t
             test-revert-flags.t
             test-revert-interactive.t
             test-revert-unknown.t
             test-revlog-ancestry.py
             test-revlog-group-emptyiter.t
             test-revlog-mmapindex.t
             test-revlog-packentry.t
             test-revlog-raw.py
             test-revlog-v2.t
             test-revset-dirstate-parents.t
             test-revset-legacy-lookup.t
             test-revset-outgoing.t
             test-rollback.t
             test-run-tests.py
             test-run-tests.t
             test-schemes.t
             test-serve.t
             test-setdiscovery.t
             test-share.t
             test-shelve.t
             test-show-stack.t
             test-show-work.t
             test-show.t
             test-simple-update.t
             test-simplekeyvaluefile.py
             test-simplemerge.py
             test-single-head.t
             test-sparse-clear.t
             test-sparse-clone.t
             test-sparse-import.t
             test-sparse-merges.t
             test-sparse-profiles.t
             test-sparse-requirement.t
             test-sparse-verbose-json.t
             test-sparse.t
             test-split.t
             test-ssh-bundle1.t
             test-ssh-clone-r.t
             test-ssh-proto-unbundle.t
             test-ssh-proto.t
             test-ssh.t
             test-sshserver.py
             test-stack.t
             test-status-inprocess.py
             test-status-rev.t
             test-status-terse.t
             test-strict.t
             test-strip-cross.t
             test-strip.t
             test-subrepo-deep-nested-change.t
             test-subrepo-missing.t
             test-subrepo-paths.t
             test-subrepo-recursion.t
             test-subrepo-relative-path.t
             test-subrepo.t
             test-symlink-os-yes-fs-no.py
             test-symlink-placeholder.t
             test-symlinks.t
             test-tag.t
             test-tags.t
             test-template-basic.t
             test-template-functions.t
             test-template-keywords.t
             test-template-map.t
             test-transplant.t
             test-treemanifest.t
             test-ui-color.py
             test-ui-config.py
             test-ui-verbosity.py
             test-unamend.t
             test-unbundlehash.t
             test-uncommit.t
             test-unified-test.t
             test-unionrepo.t
             test-unrelated-pull.t
             test-up-local-change.t
             test-update-branches.t
             test-update-dest.t
             test-update-issue1456.t
             test-update-names.t
             test-update-reverse.t
             test-upgrade-repo.t
             test-url-download.t
             test-url-rev.t
             test-url.py
             test-username-newline.t
             test-verify.t
             test-walk.t
             test-walkrepo.py
             test-websub.t
             test-win32text.t
             test-wireproto-clientreactor.py
             test-wireproto-framing.py
             test-wireproto-serverreactor.py
             test-wireproto.py
             test-wsgirequest.py
             test-xdg.t

             # bundle2.py - generic container format to transmit arbitrary data.
             #
             # Copyright 2013 Facebook, Inc.
             #
             # This software may be used and distributed according to the terms of the
             # GNU General Public License version 2 or any later version.
             """Handling of the new bundle2 format
             The goal of bundle2 is to act as an atomically packet to transmit a set of
             payloads in an application agnostic way. It consist in a sequence of "parts"
             that will be handed to and processed by the application layer.
             General format architecture
             ===========================
             The format is architectured as follow
              - magic string
              - stream level parameters
              - payload parts (any number)
              - end of stream marker.
             the Binary format
             ============================
             All numbers are unsigned and big-endian.
             stream level parameters
             ------------------------
             Binary format is as follow
             :params size: int32
               The total number of Bytes used by the parameters
             :params value: arbitrary number of Bytes
               A blob of `params size` containing the serialized version of all stream level
               parameters.
               The blob contains a space separated list of parameters. Parameters with value
               are stored in the form `<name>=<value>`. Both name and value are urlquoted.
               Empty name are obviously forbidden.
               Name MUST start with a letter. If this first letter is lower case, the
               parameter is advisory and can be safely ignored. However when the first
               letter is capital, the parameter is mandatory and the bundling process MUST
               stop if he is not able to proceed it.
               Stream parameters use a simple textual format for two main reasons:
               - Stream level parameters should remain simple and we want to discourage any
                 crazy usage.
               - Textual data allow easy human inspection of a bundle2 header in case of
                 troubles.
               Any Applicative level options MUST go into a bundle2 part instead.
             Payload part
             ------------------------
             Binary format is as follow
             :header size: int32
               The total number of Bytes used by the part header. When the header is empty
               (size = 0) this is interpreted as the end of stream marker.
             :header:
                 The header defines how to interpret the part. It contains two piece of
                 data: the part type, and the part parameters.
                 The part type is used to route an application level handler, that can
                 interpret payload.
                 Part parameters are passed to the application level handler.  They are
                 meant to convey information that will help the application level object to
                 interpret the part payload.
                 The binary format of the header is has follow
                 :typesize: (one byte)
                 :parttype: alphanumerical part name (restricted to [a-zA-Z0-9_:-]*)
                 :partid: A 32bits integer (unique in the bundle) that can be used to refer
                          to this part.
                 :parameters:
                     Part's parameter may have arbitrary content, the binary structure is::
                         <mandatory-count><advisory-count><param-sizes><param-data>
                     :mandatory-count: 1 byte, number of mandatory parameters
                     :advisory-count:  1 byte, number of advisory parameters
                     :param-sizes:
                         N couple of bytes, where N is the total number of parameters. Each
                         couple contains (<size-of-key>, <size-of-value) for one parameter.
                     :param-data:
                         A blob of bytes from which each parameter key and value can be
                         retrieved using the list of size couples stored in the previous
                         field.
                         Mandatory parameters comes first, then the advisory ones.
                         Each parameter's key MUST be unique within the part.
             :payload:
                 payload is a series of `<chunksize><chunkdata>`.
                 `chunksize` is an int32, `chunkdata` are plain bytes (as much as
                 `chunksize` says)` The payload part is concluded by a zero size chunk.
                 The current implementation always produces either zero or one chunk.
                 This is an implementation limitation that will ultimately be lifted.
                 `chunksize` can be negative to trigger special case processing. No such
                 processing is in place yet.
             Bundle processing
             ============================
             Each part is processed in order using a "part handler". Handler are registered
             for a certain part type.
             The matching of a part to its handler is case insensitive. The case of the
             part type is used to know if a part is mandatory or advisory. If the Part type
             contains any uppercase char it is considered mandatory. When no handler is
             known for a Mandatory part, the process is aborted and an exception is raised.
             If the part is advisory and no handler is known, the part is ignored. When the
             process is aborted, the full bundle is still read from the stream to keep the
             channel usable. But none of the part read from an abort are processed. In the
             future, dropping the stream may become an option for channel we do not care to
             preserve.
             """
             from __future__ import absolute_import, division
             import collections
             import errno
             import os
             import re
             import string
             import struct
             import sys
             from .i18n import _
             from . import (
                 bookmarks,
                 changegroup,
                 encoding,
                 error,
                 node as nodemod,
                 obsolete,
                 phases,
                 pushkey,
                 pycompat,
                 streamclone,
                 tags,
                 url,
                 util,
             )
             from .utils import (
                 stringutil,
             )
             urlerr = util.urlerr
             urlreq = util.urlreq
             _pack = struct.pack
             _unpack = struct.unpack
             _fstreamparamsize = '>i'
             _fpartheadersize = '>i'
             _fparttypesize = '>B'
             _fpartid = '>I'
             _fpayloadsize = '>i'
             _fpartparamcount = '>BB'
             preferedchunksize = 32768
             _parttypeforbidden = re.compile('[^a-zA-Z0-9_:-]')
             def outdebug(ui, message):
                 """debug regarding output stream (bundling)"""
                 if ui.configbool('devel', 'bundle2.debug'):
                     ui.debug('bundle2-output: %s\n' % message)
             def indebug(ui, message):
                 """debug on input stream (unbundling)"""
                 if ui.configbool('devel', 'bundle2.debug'):
                     ui.debug('bundle2-input: %s\n' % message)
             def validateparttype(parttype):
                 """raise ValueError if a parttype contains invalid character"""
                 if _parttypeforbidden.search(parttype):
                     raise ValueError(parttype)
             def _makefpartparamsizes(nbparams):
                 """return a struct format to read part parameter sizes
                 The number parameters is variable so we need to build that format
                 dynamically.
                 """
                 return '>'+('BB'*nbparams)
             parthandlermapping = {}
             def parthandler(parttype, params=()):
                 """decorator that register a function as a bundle2 part handler
                 eg::
                     @parthandler('myparttype', ('mandatory', 'param', 'handled'))
                     def myparttypehandler(...):
                         '''process a part of type "my part".'''
                         ...
                 """
                 validateparttype(parttype)
                 def _decorator(func):
                     lparttype = parttype.lower() # enforce lower case matching.
                     assert lparttype not in parthandlermapping
                     parthandlermapping[lparttype] = func
                     func.params = frozenset(params)
                     return func
                 return _decorator
             class unbundlerecords(object):
                 """keep record of what happens during and unbundle
                 New records are added using `records.add('cat', obj)`. Where 'cat' is a
                 category of record and obj is an arbitrary object.
                 `records['cat']` will return all entries of this category 'cat'.
                 Iterating on the object itself will yield `('category', obj)` tuples
                 for all entries.
                 All iterations happens in chronological order.
                 """
                 def __init__(self):
                     self._categories = {}
                     self._sequences = []
                     self._replies = {}
                 def add(self, category, entry, inreplyto=None):
                     """add a new record of a given category.
                     The entry can then be retrieved in the list returned by
                     self['category']."""
                     self._categories.setdefault(category, []).append(entry)
                     self._sequences.append((category, entry))
                     if inreplyto is not None:
                         self.getreplies(inreplyto).add(category, entry)
                 def getreplies(self, partid):
                     """get the records that are replies to a specific part"""
                     return self._replies.setdefault(partid, unbundlerecords())
                 def __getitem__(self, cat):
                     return tuple(self._categories.get(cat, ()))
                 def __iter__(self):
                     return iter(self._sequences)
                 def __len__(self):
                     return len(self._sequences)
                 def __nonzero__(self):
                     return bool(self._sequences)
                 __bool__ = __nonzero__
             class bundleoperation(object):
                 """an object that represents a single bundling process
                 Its purpose is to carry unbundle-related objects and states.
                 A new object should be created at the beginning of each bundle processing.
                 The object is to be returned by the processing function.
                 The object has very little content now it will ultimately contain:
                 * an access to the repo the bundle is applied to,
                 * a ui object,
                 * a way to retrieve a transaction to add changes to the repo,
                 * a way to record the result of processing each part,
                 * a way to construct a bundle response when applicable.
                 """
                 def __init__(self, repo, transactiongetter, captureoutput=True, source=''):
                     self.repo = repo
                     self.ui = repo.ui
                     self.records = unbundlerecords()
                     self.reply = None
                     self.captureoutput = captureoutput
                     self.hookargs = {}
                     self._gettransaction = transactiongetter
                     # carries value that can modify part behavior
                     self.modes = {}
                     self.source = source
                 def gettransaction(self):
                     transaction = self._gettransaction()
                     if self.hookargs:
                         # the ones added to the transaction supercede those added
                         # to the operation.
                         self.hookargs.update(transaction.hookargs)
                         transaction.hookargs = self.hookargs
                     # mark the hookargs as flushed.  further attempts to add to
                     # hookargs will result in an abort.
                     self.hookargs = None
                     return transaction
                 def addhookargs(self, hookargs):
                     if self.hookargs is None:
                         raise error.ProgrammingError('attempted to add hookargs to '
                                                      'operation after transaction started')
                     self.hookargs.update(hookargs)
             class TransactionUnavailable(RuntimeError):
                 pass
             def _notransaction():
                 """default method to get a transaction while processing a bundle
                 Raise an exception to highlight the fact that no transaction was expected
                 to be created"""
                 raise TransactionUnavailable()
             def applybundle(repo, unbundler, tr, source, url=None, **kwargs):
                 # transform me into unbundler.apply() as soon as the freeze is lifted
                 if isinstance(unbundler, unbundle20):
                     tr.hookargs['bundle2'] = '1'
                     if source is not None and 'source' not in tr.hookargs:
                         tr.hookargs['source'] = source
                     if url is not None and 'url' not in tr.hookargs:
                         tr.hookargs['url'] = url
                     return processbundle(repo, unbundler, lambda: tr, source=source)
                 else:
                     # the transactiongetter won't be used, but we might as well set it
                     op = bundleoperation(repo, lambda: tr, source=source)
                     _processchangegroup(op, unbundler, tr, source, url, **kwargs)
                     return op
             class partiterator(object):
                 def __init__(self, repo, op, unbundler):
                     self.repo = repo
                     self.op = op
                     self.unbundler = unbundler
                     self.iterator = None
                     self.count = 0
                     self.current = None
                 def __enter__(self):
                     def func():
                         itr = enumerate(self.unbundler.iterparts())
                         for count, p in itr:
                             self.count = count
                             self.current = p
                             yield p
                             p.consume()
                             self.current = None
                     self.iterator = func()
                     return self.iterator
                 def __exit__(self, type, exc, tb):
                     if not self.iterator:
                         return
                     # Only gracefully abort in a normal exception situation. User aborts
                     # like Ctrl+C throw a KeyboardInterrupt which is not a base Exception,
                     # and should not gracefully cleanup.
                     if isinstance(exc, Exception):
                         # Any exceptions seeking to the end of the bundle at this point are
                         # almost certainly related to the underlying stream being bad.
                         # And, chances are that the exception we're handling is related to
                         # getting in that bad state. So, we swallow the seeking error and
                         # re-raise the original error.
                         seekerror = False
                         try:
                             if self.current:
                                 # consume the part content to not corrupt the stream.
                                 self.current.consume()
                             for part in self.iterator:
                                 # consume the bundle content
                                 part.consume()
                         except Exception:
                             seekerror = True
                         # Small hack to let caller code distinguish exceptions from bundle2
                         # processing from processing the old format. This is mostly needed
                         # to handle different return codes to unbundle according to the type
                         # of bundle. We should probably clean up or drop this return code
                         # craziness in a future version.
                         exc.duringunbundle2 = True
                         salvaged = []
                         replycaps = None
                         if self.op.reply is not None:
                             salvaged = self.op.reply.salvageoutput()
                             replycaps = self.op.reply.capabilities
                         exc._replycaps = replycaps
                         exc._bundle2salvagedoutput = salvaged
                         # Re-raising from a variable loses the original stack. So only use
                         # that form if we need to.
                         if seekerror:
                             raise exc
                     self.repo.ui.debug('bundle2-input-bundle: %i parts total\n' %
                                        self.count)
             def processbundle(repo, unbundler, transactiongetter=None, op=None, source=''):
                 """This function process a bundle, apply effect to/from a repo
                 It iterates over each part then searches for and uses the proper handling
                 code to process the part. Parts are processed in order.
                 Unknown Mandatory part will abort the process.
                 It is temporarily possible to provide a prebuilt bundleoperation to the
                 function. This is used to ensure output is properly propagated in case of
                 an error during the unbundling. This output capturing part will likely be
                 reworked and this ability will probably go away in the process.
                 """
                 if op is None:
                     if transactiongetter is None:
                         transactiongetter = _notransaction
                     op = bundleoperation(repo, transactiongetter, source=source)
                 # todo:
                 # - replace this is a init function soon.
                 # - exception catching
                 unbundler.params
                 if repo.ui.debugflag:
                     msg = ['bundle2-input-bundle:']
                     if unbundler.params:
                         msg.append(' %i params' % len(unbundler.params))
                     if op._gettransaction is None or op._gettransaction is _notransaction:
                         msg.append(' no-transaction')
                     else:
                         msg.append(' with-transaction')
                     msg.append('\n')
                     repo.ui.debug(''.join(msg))
                 processparts(repo, op, unbundler)
                 return op
             def processparts(repo, op, unbundler):
                 with partiterator(repo, op, unbundler) as parts:
                     for part in parts:
                         _processpart(op, part)
             def _processchangegroup(op, cg, tr, source, url, **kwargs):
                 ret = cg.apply(op.repo, tr, source, url, **kwargs)
                 op.records.add('changegroup', {
                     'return': ret,
                 })
                 return ret
             def _gethandler(op, part):
                 status = 'unknown' # used by debug output
                 try:
                     handler = parthandlermapping.get(part.type)
                     if handler is None:
                         status = 'unsupported-type'
                         raise error.BundleUnknownFeatureError(parttype=part.type)
                     indebug(op.ui, 'found a handler for part %s' % part.type)
                     unknownparams = part.mandatorykeys - handler.params
                     if unknownparams:
                         unknownparams = list(unknownparams)
                         unknownparams.sort()
                         status = 'unsupported-params (%s)' % ', '.join(unknownparams)
                         raise error.BundleUnknownFeatureError(parttype=part.type,
                                                               params=unknownparams)
                     status = 'supported'
                 except error.BundleUnknownFeatureError as exc:
                     if part.mandatory: # mandatory parts
                         raise
                     indebug(op.ui, 'ignoring unsupported advisory part %s' % exc)
                     return # skip to part processing
                 finally:
                     if op.ui.debugflag:
                         msg = ['bundle2-input-part: "%s"' % part.type]
                         if not part.mandatory:
                             msg.append(' (advisory)')
                         nbmp = len(part.mandatorykeys)
                         nbap = len(part.params) - nbmp
                         if nbmp or nbap:
                             msg.append(' (params:')
                             if nbmp:
                                 msg.append(' %i mandatory' % nbmp)
                             if nbap:
                                 msg.append(' %i advisory' % nbmp)
                             msg.append(')')
                         msg.append(' %s\n' % status)
                         op.ui.debug(''.join(msg))
                 return handler
             def _processpart(op, part):
                 """process a single part from a bundle
                 The part is guaranteed to have been fully consumed when the function exits
                 (even if an exception is raised)."""
                 handler = _gethandler(op, part)
                 if handler is None:
                     return
                 # handler is called outside the above try block so that we don't
                 # risk catching KeyErrors from anything other than the
                 # parthandlermapping lookup (any KeyError raised by handler()
                 # itself represents a defect of a different variety).
                 output = None
                 if op.captureoutput and op.reply is not None:
                     op.ui.pushbuffer(error=True, subproc=True)
                     output = ''
                 try:
                     handler(op, part)
                 finally:
                     if output is not None:
                         output = op.ui.popbuffer()
                     if output:
                         outpart = op.reply.newpart('output', data=output,
                                                    mandatory=False)
                         outpart.addparam(
                             'in-reply-to', pycompat.bytestr(part.id), mandatory=False)
             def decodecaps(blob):
                 """decode a bundle2 caps bytes blob into a dictionary
                 The blob is a list of capabilities (one per line)
                 Capabilities may have values using a line of the form::
                     capability=value1,value2,value3
                 The values are always a list."""
                 caps = {}
                 for line in blob.splitlines():
                     if not line:
                         continue
                     if '=' not in line:
                         key, vals = line, ()
                     else:
                         key, vals = line.split('=', 1)
                         vals = vals.split(',')
                     key = urlreq.unquote(key)
                     vals = [urlreq.unquote(v) for v in vals]
                     caps[key] = vals
                 return caps
             def encodecaps(caps):
                 """encode a bundle2 caps dictionary into a bytes blob"""
                 chunks = []
                 for ca in sorted(caps):
                     vals = caps[ca]
                     ca = urlreq.quote(ca)
                     vals = [urlreq.quote(v) for v in vals]
                     if vals:
                         ca = "%s=%s" % (ca, ','.join(vals))
                     chunks.append(ca)
                 return '\n'.join(chunks)
             bundletypes = {
                 "": ("", 'UN'),       # only when using unbundle on ssh and old http servers
                                       # since the unification ssh accepts a header but there
                                       # is no capability signaling it.
                 "HG20": (), # special-cased below
                 "HG10UN": ("HG10UN", 'UN'),
                 "HG10BZ": ("HG10", 'BZ'),
                 "HG10GZ": ("HG10GZ", 'GZ'),
             }
             # hgweb uses this list to communicate its preferred type
             bundlepriority = ['HG10GZ', 'HG10BZ', 'HG10UN']
             class bundle20(object):
                 """represent an outgoing bundle2 container
                 Use the `addparam` method to add stream level parameter. and `newpart` to
                 populate it. Then call `getchunks` to retrieve all the binary chunks of
                 data that compose the bundle2 container."""
                 _magicstring = 'HG20'
                 def __init__(self, ui, capabilities=()):
                     self.ui = ui
                     self._params = []
                     self._parts = []
                     self.capabilities = dict(capabilities)
                     self._compengine = util.compengines.forbundletype('UN')
                     self._compopts = None
                     # If compression is being handled by a consumer of the raw
                     # data (e.g. the wire protocol), unsetting this flag tells
                     # consumers that the bundle is best left uncompressed.
                     self.prefercompressed = True
                 def setcompression(self, alg, compopts=None):
                     """setup core part compression to <alg>"""
                     if alg in (None, 'UN'):
                         return
                     assert not any(n.lower() == 'compression' for n, v in self._params)
                     self.addparam('Compression', alg)
                     self._compengine = util.compengines.forbundletype(alg)
                     self._compopts = compopts
                 @property
                 def nbparts(self):
                     """total number of parts added to the bundler"""
                     return len(self._parts)
                 # methods used to defines the bundle2 content
                 def addparam(self, name, value=None):
                     """add a stream level parameter"""
                     if not name:
                         raise error.ProgrammingError(b'empty parameter name')
                     if name[0:1] not in pycompat.bytestr(string.ascii_letters):
                         raise error.ProgrammingError(b'non letter first character: %s'
                                                      % name)
                     self._params.append((name, value))
                 def addpart(self, part):
                     """add a new part to the bundle2 container
                     Parts contains the actual applicative payload."""
                     assert part.id is None
                     part.id = len(self._parts) # very cheap counter
                     self._parts.append(part)
                 def newpart(self, typeid, *args, **kwargs):
                     """create a new part and add it to the containers
                     As the part is directly added to the containers. For now, this means
                     that any failure to properly initialize the part after calling
                     ``newpart`` should result in a failure of the whole bundling process.
                     You can still fall back to manually create and add if you need better
                     control."""
                     part = bundlepart(typeid, *args, **kwargs)
                     self.addpart(part)
                     return part
                 # methods used to generate the bundle2 stream
                 def getchunks(self):
                     if self.ui.debugflag:
                         msg = ['bundle2-output-bundle: "%s",' % self._magicstring]
                         if self._params:
                             msg.append(' (%i params)' % len(self._params))
                         msg.append(' %i parts total\n' % len(self._parts))
                         self.ui.debug(''.join(msg))
                     outdebug(self.ui, 'start emission of %s stream' % self._magicstring)
                     yield self._magicstring
                     param = self._paramchunk()
                     outdebug(self.ui, 'bundle parameter: %s' % param)
                     yield _pack(_fstreamparamsize, len(param))
                     if param:
                         yield param
                     for chunk in self._compengine.compressstream(self._getcorechunk(),
                                                                  self._compopts):
                         yield chunk
                 def _paramchunk(self):
                     """return a encoded version of all stream parameters"""
                     blocks = []
                     for par, value in self._params:
                         par = urlreq.quote(par)
                         if value is not None:
                             value = urlreq.quote(value)
                             par = '%s=%s' % (par, value)
                         blocks.append(par)
                     return ' '.join(blocks)
                 def _getcorechunk(self):
                     """yield chunk for the core part of the bundle
                     (all but headers and parameters)"""
                     outdebug(self.ui, 'start of parts')
                     for part in self._parts:
                         outdebug(self.ui, 'bundle part: "%s"' % part.type)
                         for chunk in part.getchunks(ui=self.ui):
                             yield chunk
                     outdebug(self.ui, 'end of bundle')
                     yield _pack(_fpartheadersize, 0)
                 def salvageoutput(self):
                     """return a list with a copy of all output parts in the bundle
                     This is meant to be used during error handling to make sure we preserve
                     server output"""
                     salvaged = []
                     for part in self._parts:
                         if part.type.startswith('output'):
                             salvaged.append(part.copy())
                     return salvaged
             class unpackermixin(object):
                 """A mixin to extract bytes and struct data from a stream"""
                 def __init__(self, fp):
                     self._fp = fp
                 def _unpack(self, format):
                     """unpack this struct format from the stream
                     This method is meant for internal usage by the bundle2 protocol only.
                     They directly manipulate the low level stream including bundle2 level
                     instruction.
                     Do not use it to implement higher-level logic or methods."""
                     data = self._readexact(struct.calcsize(format))
                     return _unpack(format, data)
                 def _readexact(self, size):
                     """read exactly <size> bytes from the stream
                     This method is meant for internal usage by the bundle2 protocol only.
                     They directly manipulate the low level stream including bundle2 level
                     instruction.
                     Do not use it to implement higher-level logic or methods."""
                     return changegroup.readexactly(self._fp, size)
             def getunbundler(ui, fp, magicstring=None):
                 """return a valid unbundler object for a given magicstring"""
                 if magicstring is None:
                     magicstring = changegroup.readexactly(fp, 4)
                 magic, version = magicstring[0:2], magicstring[2:4]
                 if magic != 'HG':
                     ui.debug(
                         "error: invalid magic: %r (version %r), should be 'HG'\n"
                         % (magic, version))
                     raise error.Abort(_('not a Mercurial bundle'))
                 unbundlerclass = formatmap.get(version)
                 if unbundlerclass is None:
                     raise error.Abort(_('unknown bundle version %s') % version)
                 unbundler = unbundlerclass(ui, fp)
                 indebug(ui, 'start processing of %s stream' % magicstring)
                 return unbundler
             class unbundle20(unpackermixin):
                 """interpret a bundle2 stream
                 This class is fed with a binary stream and yields parts through its
                 `iterparts` methods."""
                 _magicstring = 'HG20'
                 def __init__(self, ui, fp):
                     """If header is specified, we do not read it out of the stream."""
                     self.ui = ui
                     self._compengine = util.compengines.forbundletype('UN')
                     self._compressed = None
                     super(unbundle20, self).__init__(fp)
                 @util.propertycache
                 def params(self):
                     """dictionary of stream level parameters"""
                     indebug(self.ui, 'reading bundle2 stream parameters')
                     params = {}
                     paramssize = self._unpack(_fstreamparamsize)[0]
                     if paramssize < 0:
                         raise error.BundleValueError('negative bundle param size: %i'
                                                      % paramssize)
                     if paramssize:
                         params = self._readexact(paramssize)
                         params = self._processallparams(params)
                     return params
                 def _processallparams(self, paramsblock):
                     """"""
                     params = util.sortdict()
                     for p in paramsblock.split(' '):
                         p = p.split('=', 1)
                         p = [urlreq.unquote(i) for i in p]
                         if len(p) < 2:
                             p.append(None)
                         self._processparam(*p)
                         params[p[0]] = p[1]
                     return params
                 def _processparam(self, name, value):
                     """process a parameter, applying its effect if needed
                     Parameter starting with a lower case letter are advisory and will be
                     ignored when unknown.  Those starting with an upper case letter are
                     mandatory and will this function will raise a KeyError when unknown.
                     Note: no option are currently supported. Any input will be either
                           ignored or failing.
                     """
                     if not name:
                         raise ValueError(r'empty parameter name')
                     if name[0:1] not in pycompat.bytestr(string.ascii_letters):
                         raise ValueError(r'non letter first character: %s' % name)
                     try:
                         handler = b2streamparamsmap[name.lower()]
                     except KeyError:
                         if name[0:1].islower():
                             indebug(self.ui, "ignoring unknown parameter %s" % name)
                         else:
                             raise error.BundleUnknownFeatureError(params=(name,))
                     else:
                         handler(self, name, value)
                 def _forwardchunks(self):
                     """utility to transfer a bundle2 as binary
                     This is made necessary by the fact the 'getbundle' command over 'ssh'
                     have no way to know then the reply end, relying on the bundle to be
                     interpreted to know its end. This is terrible and we are sorry, but we
                     needed to move forward to get general delta enabled.
                     """
                     yield self._magicstring
                     assert 'params' not in vars(self)
                     paramssize = self._unpack(_fstreamparamsize)[0]
                     if paramssize < 0:
                         raise error.BundleValueError('negative bundle param size: %i'
                                                      % paramssize)
                     yield _pack(_fstreamparamsize, paramssize)
                     if paramssize:
                         params = self._readexact(paramssize)
                         self._processallparams(params)
                         yield params
                         assert self._compengine.bundletype == 'UN'
                     # From there, payload might need to be decompressed
                     self._fp = self._compengine.decompressorreader(self._fp)
                     emptycount = 0
                     while emptycount < 2:
                         # so we can brainlessly loop
                         assert _fpartheadersize == _fpayloadsize
                         size = self._unpack(_fpartheadersize)[0]
                         yield _pack(_fpartheadersize, size)
                         if size:
                             emptycount = 0
                         else:
                             emptycount += 1
                             continue
                         if size == flaginterrupt:
                             continue
                         elif size < 0:
                             raise error.BundleValueError('negative chunk size: %i')
                         yield self._readexact(size)
                 def iterparts(self, seekable=False):
                     """yield all parts contained in the stream"""
                     cls = seekableunbundlepart if seekable else unbundlepart
                     # make sure param have been loaded
                     self.params
                     # From there, payload need to be decompressed
                     self._fp = self._compengine.decompressorreader(self._fp)
                     indebug(self.ui, 'start extraction of bundle2 parts')
                     headerblock = self._readpartheader()
                     while headerblock is not None:
                         part = cls(self.ui, headerblock, self._fp)
                         yield part
                         # Ensure part is fully consumed so we can start reading the next
                         # part.
                         part.consume()
                         headerblock = self._readpartheader()
                     indebug(self.ui, 'end of bundle2 stream')
                 def _readpartheader(self):
                     """reads a part header size and return the bytes blob
                     returns None if empty"""
                     headersize = self._unpack(_fpartheadersize)[0]
                     if headersize < 0:
                         raise error.BundleValueError('negative part header size: %i'
                                                      % headersize)
                     indebug(self.ui, 'part header size: %i' % headersize)
                     if headersize:
                         return self._readexact(headersize)
                     return None
                 def compressed(self):
                     self.params # load params
                     return self._compressed
                 def close(self):
                     """close underlying file"""
                     if util.safehasattr(self._fp, 'close'):
                         return self._fp.close()
             formatmap = {'20': unbundle20}
             b2streamparamsmap = {}
             def b2streamparamhandler(name):
                 """register a handler for a stream level parameter"""
                 def decorator(func):
                     assert name not in formatmap
                     b2streamparamsmap[name] = func
                     return func
                 return decorator
             @b2streamparamhandler('compression')
             def processcompression(unbundler, param, value):
                 """read compression parameter and install payload decompression"""
                 if value not in util.compengines.supportedbundletypes:
                     raise error.BundleUnknownFeatureError(params=(param,),
                                                           values=(value,))
                 unbundler._compengine = util.compengines.forbundletype(value)
                 if value is not None:
                     unbundler._compressed = True
             class bundlepart(object):
                 """A bundle2 part contains application level payload
                 The part `type` is used to route the part to the application level
                 handler.
                 The part payload is contained in ``part.data``. It could be raw bytes or a
                 generator of byte chunks.
                 You can add parameters to the part using the ``addparam`` method.
                 Parameters can be either mandatory (default) or advisory. Remote side
                 should be able to safely ignore the advisory ones.
                 Both data and parameters cannot be modified after the generation has begun.
                 """
                 def __init__(self, parttype, mandatoryparams=(), advisoryparams=(),
                              data='', mandatory=True):
                     validateparttype(parttype)
                     self.id = None
                     self.type = parttype
                     self._data = data
                     self._mandatoryparams = list(mandatoryparams)
                     self._advisoryparams = list(advisoryparams)
                     # checking for duplicated entries
                     self._seenparams = set()
                     for pname, __ in self._mandatoryparams + self._advisoryparams:
                         if pname in self._seenparams:
                             raise error.ProgrammingError('duplicated params: %s' % pname)
                         self._seenparams.add(pname)
                     # status of the part's generation:
                     # - None: not started,
                     # - False: currently generated,
                     # - True: generation done.
                     self._generated = None
                     self.mandatory = mandatory
                 def __repr__(self):
                     cls = "%s.%s" % (self.__class__.__module__, self.__class__.__name__)
                     return ('<%s object at %x; id: %s; type: %s; mandatory: %s>'
                             % (cls, id(self), self.id, self.type, self.mandatory))
                 def copy(self):
                     """return a copy of the part
                     The new part have the very same content but no partid assigned yet.
                     Parts with generated data cannot be copied."""
                     assert not util.safehasattr(self.data, 'next')
                     return self.__class__(self.type, self._mandatoryparams,
                                           self._advisoryparams, self._data, self.mandatory)
                 # methods used to defines the part content
                 @property
                 def data(self):
                     return self._data
                 @data.setter
                 def data(self, data):
                     if self._generated is not None:
                         raise error.ReadOnlyPartError('part is being generated')
                     self._data = data
                 @property
                 def mandatoryparams(self):
                     # make it an immutable tuple to force people through ``addparam``
                     return tuple(self._mandatoryparams)
                 @property
                 def advisoryparams(self):
                     # make it an immutable tuple to force people through ``addparam``
                     return tuple(self._advisoryparams)
                 def addparam(self, name, value='', mandatory=True):
                     """add a parameter to the part
                     If 'mandatory' is set to True, the remote handler must claim support
                     for this parameter or the unbundling will be aborted.
                     The 'name' and 'value' cannot exceed 255 bytes each.
                     """
                     if self._generated is not None:
                         raise error.ReadOnlyPartError('part is being generated')
                     if name in self._seenparams:
                         raise ValueError('duplicated params: %s' % name)
                     self._seenparams.add(name)
                     params = self._advisoryparams
                     if mandatory:
                         params = self._mandatoryparams
                     params.append((name, value))
                 # methods used to generates the bundle2 stream
                 def getchunks(self, ui):
                     if self._generated is not None:
                         raise error.ProgrammingError('part can only be consumed once')
                     self._generated = False
                     if ui.debugflag:
                         msg = ['bundle2-output-part: "%s"' % self.type]
                         if not self.mandatory:
                             msg.append(' (advisory)')
                         nbmp = len(self.mandatoryparams)
                         nbap = len(self.advisoryparams)
                         if nbmp or nbap:
                             msg.append(' (params:')
                             if nbmp:
                                 msg.append(' %i mandatory' % nbmp)
                             if nbap:
                                 msg.append(' %i advisory' % nbmp)
                             msg.append(')')
                         if not self.data:
                             msg.append(' empty payload')
                         elif (util.safehasattr(self.data, 'next')
                               or util.safehasattr(self.data, '__next__')):
                             msg.append(' streamed payload')
                         else:
                             msg.append(' %i bytes payload' % len(self.data))
                         msg.append('\n')
                         ui.debug(''.join(msg))
                     #### header
                     if self.mandatory:
                         parttype = self.type.upper()
                     else:
                         parttype = self.type.lower()
                     outdebug(ui, 'part %s: "%s"' % (pycompat.bytestr(self.id), parttype))
                     ## parttype
                     header = [_pack(_fparttypesize, len(parttype)),
                               parttype, _pack(_fpartid, self.id),
                              ]
                     ## parameters
                     # count
                     manpar = self.mandatoryparams
                     advpar = self.advisoryparams
                     header.append(_pack(_fpartparamcount, len(manpar), len(advpar)))
                     # size
                     parsizes = []
                     for key, value in manpar:
                         parsizes.append(len(key))
                         parsizes.append(len(value))
                     for key, value in advpar:
                         parsizes.append(len(key))
                         parsizes.append(len(value))
                     paramsizes = _pack(_makefpartparamsizes(len(parsizes) // 2), *parsizes)
                     header.append(paramsizes)
                     # key, value
                     for key, value in manpar:
                         header.append(key)
                         header.append(value)
                     for key, value in advpar:
                         header.append(key)
                         header.append(value)
                     ## finalize header
                     try:
                         headerchunk = ''.join(header)
                     except TypeError:
                         raise TypeError(r'Found a non-bytes trying to '
                                         r'build bundle part header: %r' % header)
                     outdebug(ui, 'header chunk size: %i' % len(headerchunk))
                     yield _pack(_fpartheadersize, len(headerchunk))
                     yield headerchunk
                     ## payload
                     try:
                         for chunk in self._payloadchunks():
                             outdebug(ui, 'payload chunk size: %i' % len(chunk))
                             yield _pack(_fpayloadsize, len(chunk))
                             yield chunk
                     except GeneratorExit:
                         # GeneratorExit means that nobody is listening for our
                         # results anyway, so just bail quickly rather than trying
                         # to produce an error part.
                         ui.debug('bundle2-generatorexit\n')
                         raise
                     except BaseException as exc:
                         bexc = stringutil.forcebytestr(exc)
                         # backup exception data for later
                         ui.debug('bundle2-input-stream-interrupt: encoding exception %s'
                                  % bexc)
                         tb = sys.exc_info()[2]
                         msg = 'unexpected error: %s' % bexc
                         interpart = bundlepart('error:abort', [('message', msg)],
                                                mandatory=False)
                         interpart.id = 0
                         yield _pack(_fpayloadsize, -1)
                         for chunk in interpart.getchunks(ui=ui):
                             yield chunk
                         outdebug(ui, 'closing payload chunk')
                         # abort current part payload
                         yield _pack(_fpayloadsize, 0)
                         pycompat.raisewithtb(exc, tb)
                     # end of payload
                     outdebug(ui, 'closing payload chunk')
                     yield _pack(_fpayloadsize, 0)
                     self._generated = True
                 def _payloadchunks(self):
                     """yield chunks of a the part payload
                     Exists to handle the different methods to provide data to a part."""
                     # we only support fixed size data now.
                     # This will be improved in the future.
                     if (util.safehasattr(self.data, 'next')
                         or util.safehasattr(self.data, '__next__')):
                         buff = util.chunkbuffer(self.data)
                         chunk = buff.read(preferedchunksize)
                         while chunk:
                             yield chunk
                             chunk = buff.read(preferedchunksize)
                     elif len(self.data):
                         yield self.data
             flaginterrupt = -1
             class interrupthandler(unpackermixin):
                 """read one part and process it with restricted capability
                 This allows to transmit exception raised on the producer size during part
                 iteration while the consumer is reading a part.
                 Part processed in this manner only have access to a ui object,"""
                 def __init__(self, ui, fp):
                     super(interrupthandler, self).__init__(fp)
                     self.ui = ui
                 def _readpartheader(self):
                     """reads a part header size and return the bytes blob
                     returns None if empty"""
                     headersize = self._unpack(_fpartheadersize)[0]
                     if headersize < 0:
                         raise error.BundleValueError('negative part header size: %i'
                                                      % headersize)
                     indebug(self.ui, 'part header size: %i\n' % headersize)
                     if headersize:
                         return self._readexact(headersize)
                     return None
                 def __call__(self):
                     self.ui.debug('bundle2-input-stream-interrupt:'
                                   ' opening out of band context\n')
                     indebug(self.ui, 'bundle2 stream interruption, looking for a part.')
                     headerblock = self._readpartheader()
                     if headerblock is None:
                         indebug(self.ui, 'no part found during interruption.')
                         return
                     part = unbundlepart(self.ui, headerblock, self._fp)
                     op = interruptoperation(self.ui)
                     hardabort = False
                     try:
                         _processpart(op, part)
                     except (SystemExit, KeyboardInterrupt):
                         hardabort = True
                         raise
                     finally:
                         if not hardabort:
                             part.consume()
                     self.ui.debug('bundle2-input-stream-interrupt:'
                                   ' closing out of band context\n')
             class interruptoperation(object):
                 """A limited operation to be use by part handler during interruption
                 It only have access to an ui object.
                 """
                 def __init__(self, ui):
                     self.ui = ui
                     self.reply = None
                     self.captureoutput = False
                 @property
                 def repo(self):
                     raise error.ProgrammingError('no repo access from stream interruption')
                 def gettransaction(self):
                     raise TransactionUnavailable('no repo access from stream interruption')
             def decodepayloadchunks(ui, fh):
                 """Reads bundle2 part payload data into chunks.
                 Part payload data consists of framed chunks. This function takes
                 a file handle and emits those chunks.
                 """
                 dolog = ui.configbool('devel', 'bundle2.debug')
                 debug = ui.debug
                 headerstruct = struct.Struct(_fpayloadsize)
                 headersize = headerstruct.size
                 unpack = headerstruct.unpack
                 readexactly = changegroup.readexactly
                 read = fh.read
                 chunksize = unpack(readexactly(fh, headersize))[0]
                 indebug(ui, 'payload chunk size: %i' % chunksize)
                 # changegroup.readexactly() is inlined below for performance.
                 while chunksize:
                     if chunksize >= 0:
                         s = read(chunksize)
                         if len(s) < chunksize:
                             raise error.Abort(_('stream ended unexpectedly '
                                                 ' (got %d bytes, expected %d)') %
                                               (len(s), chunksize))
                         yield s
                     elif chunksize == flaginterrupt:
                         # Interrupt "signal" detected. The regular stream is interrupted
                         # and a bundle2 part follows. Consume it.
                         interrupthandler(ui, fh)()
                     else:
                         raise error.BundleValueError(
                             'negative payload chunk size: %s' % chunksize)
                     s = read(headersize)
                     if len(s) < headersize:
                         raise error.Abort(_('stream ended unexpectedly '
                                             ' (got %d bytes, expected %d)') %
                                           (len(s), chunksize))
                     chunksize = unpack(s)[0]
                     # indebug() inlined for performance.
                     if dolog:
                         debug('bundle2-input: payload chunk size: %i\n' % chunksize)
             class unbundlepart(unpackermixin):
                 """a bundle part read from a bundle"""
                 def __init__(self, ui, header, fp):
                     super(unbundlepart, self).__init__(fp)
                     self._seekable = (util.safehasattr(fp, 'seek') and
                                       util.safehasattr(fp, 'tell'))
                     self.ui = ui
                     # unbundle state attr
                     self._headerdata = header
                     self._headeroffset = 0
                     self._initialized = False
                     self.consumed = False
                     # part data
                     self.id = None
                     self.type = None
                     self.mandatoryparams = None
                     self.advisoryparams = None
                     self.params = None
                     self.mandatorykeys = ()
                     self._readheader()
                     self._mandatory = None
                     self._pos = 0
                 def _fromheader(self, size):
                     """return the next <size> byte from the header"""
                     offset = self._headeroffset
                     data = self._headerdata[offset:(offset + size)]
                     self._headeroffset = offset + size
                     return data
                 def _unpackheader(self, format):
                     """read given format from header
                     This automatically compute the size of the format to read."""
                     data = self._fromheader(struct.calcsize(format))
                     return _unpack(format, data)
                 def _initparams(self, mandatoryparams, advisoryparams):
                     """internal function to setup all logic related parameters"""
                     # make it read only to prevent people touching it by mistake.
                     self.mandatoryparams = tuple(mandatoryparams)
                     self.advisoryparams  = tuple(advisoryparams)
                     # user friendly UI
                     self.params = util.sortdict(self.mandatoryparams)
                     self.params.update(self.advisoryparams)
                     self.mandatorykeys = frozenset(p[0] for p in mandatoryparams)
                 def _readheader(self):
                     """read the header and setup the object"""
                     typesize = self._unpackheader(_fparttypesize)[0]
                     self.type = self._fromheader(typesize)
                     indebug(self.ui, 'part type: "%s"' % self.type)
                     self.id = self._unpackheader(_fpartid)[0]
                     indebug(self.ui, 'part id: "%s"' % pycompat.bytestr(self.id))
                     # extract mandatory bit from type
                     self.mandatory = (self.type != self.type.lower())
                     self.type = self.type.lower()
                     ## reading parameters
                     # param count
                     mancount, advcount = self._unpackheader(_fpartparamcount)
                     indebug(self.ui, 'part parameters: %i' % (mancount + advcount))
                     # param size
                     fparamsizes = _makefpartparamsizes(mancount + advcount)
                     paramsizes = self._unpackheader(fparamsizes)
                     # make it a list of couple again
                     paramsizes = list(zip(paramsizes[::2], paramsizes[1::2]))
                     # split mandatory from advisory
                     mansizes = paramsizes[:mancount]
                     advsizes = paramsizes[mancount:]
                     # retrieve param value
                     manparams = []
                     for key, value in mansizes:
                         manparams.append((self._fromheader(key), self._fromheader(value)))
                     advparams = []
                     for key, value in advsizes:
                         advparams.append((self._fromheader(key), self._fromheader(value)))
                     self._initparams(manparams, advparams)
                     ## part payload
                     self._payloadstream = util.chunkbuffer(self._payloadchunks())
                     # we read the data, tell it
                     self._initialized = True
                 def _payloadchunks(self):
                     """Generator of decoded chunks in the payload."""
                     return decodepayloadchunks(self.ui, self._fp)
                 def consume(self):
                     """Read the part payload until completion.
                     By consuming the part data, the underlying stream read offset will
                     be advanced to the next part (or end of stream).
                     """
                     if self.consumed:
                         return
                     chunk = self.read(32768)
                     while chunk:
                         self._pos += len(chunk)
                         chunk = self.read(32768)
                 def read(self, size=None):
                     """read payload data"""
                     if not self._initialized:
                         self._readheader()
                     if size is None:
                         data = self._payloadstream.read()
                     else:
                         data = self._payloadstream.read(size)
                     self._pos += len(data)
                     if size is None or len(data) < size:
                         if not self.consumed and self._pos:
                             self.ui.debug('bundle2-input-part: total payload size %i\n'
                                           % self._pos)
                         self.consumed = True
                     return data
             class seekableunbundlepart(unbundlepart):
                 """A bundle2 part in a bundle that is seekable.
                 Regular ``unbundlepart`` instances can only be read once. This class
                 extends ``unbundlepart`` to enable bi-directional seeking within the
                 part.
                 Bundle2 part data consists of framed chunks. Offsets when seeking
                 refer to the decoded data, not the offsets in the underlying bundle2
                 stream.
                 To facilitate quickly seeking within the decoded data, instances of this
                 class maintain a mapping between offsets in the underlying stream and
                 the decoded payload. This mapping will consume memory in proportion
                 to the number of chunks within the payload (which almost certainly
                 increases in proportion with the size of the part).
                 """
                 def __init__(self, ui, header, fp):
                     # (payload, file) offsets for chunk starts.
                     self._chunkindex = []
                     super(seekableunbundlepart, self).__init__(ui, header, fp)
                 def _payloadchunks(self, chunknum=0):
                     '''seek to specified chunk and start yielding data'''
                     if len(self._chunkindex) == 0:
                         assert chunknum == 0, 'Must start with chunk 0'
                         self._chunkindex.append((0, self._tellfp()))
                     else:
                         assert chunknum < len(self._chunkindex), \
                                'Unknown chunk %d' % chunknum
                         self._seekfp(self._chunkindex[chunknum][1])
                     pos = self._chunkindex[chunknum][0]
                     for chunk in decodepayloadchunks(self.ui, self._fp):
                         chunknum += 1
                         pos += len(chunk)
                         if chunknum == len(self._chunkindex):
                             self._chunkindex.append((pos, self._tellfp()))
                         yield chunk
                 def _findchunk(self, pos):
                     '''for a given payload position, return a chunk number and offset'''
                     for chunk, (ppos, fpos) in enumerate(self._chunkindex):
                         if ppos == pos:
                             return chunk, 0
                         elif ppos > pos:
                             return chunk - 1, pos - self._chunkindex[chunk - 1][0]
                     raise ValueError('Unknown chunk')
                 def tell(self):
                     return self._pos
                 def seek(self, offset, whence=os.SEEK_SET):
                     if whence == os.SEEK_SET:
                         newpos = offset
                     elif whence == os.SEEK_CUR:
                         newpos = self._pos + offset
                     elif whence == os.SEEK_END:
                         if not self.consumed:
                             # Can't use self.consume() here because it advances self._pos.
                             chunk = self.read(32768)
                             while chunk:
                                 chunk = self.read(32768)
                         newpos = self._chunkindex[-1][0] - offset
                     else:
                         raise ValueError('Unknown whence value: %r' % (whence,))
                     if newpos > self._chunkindex[-1][0] and not self.consumed:
                         # Can't use self.consume() here because it advances self._pos.
                         chunk = self.read(32768)
                         while chunk:
                             chunk = self.read(32668)
                     if not 0 <= newpos <= self._chunkindex[-1][0]:
                         raise ValueError('Offset out of range')
                     if self._pos != newpos:
                         chunk, internaloffset = self._findchunk(newpos)
                         self._payloadstream = util.chunkbuffer(self._payloadchunks(chunk))
                         adjust = self.read(internaloffset)
                         if len(adjust) != internaloffset:
                             raise error.Abort(_('Seek failed\n'))
                         self._pos = newpos
                 def _seekfp(self, offset, whence=0):
                     """move the underlying file pointer
                     This method is meant for internal usage by the bundle2 protocol only.
                     They directly manipulate the low level stream including bundle2 level
                     instruction.
                     Do not use it to implement higher-level logic or methods."""
                     if self._seekable:
                         return self._fp.seek(offset, whence)
                     else:
                         raise NotImplementedError(_('File pointer is not seekable'))
                 def _tellfp(self):
                     """return the file offset, or None if file is not seekable
                     This method is meant for internal usage by the bundle2 protocol only.
                     They directly manipulate the low level stream including bundle2 level
                     instruction.
                     Do not use it to implement higher-level logic or methods."""
                     if self._seekable:
                         try:
                             return self._fp.tell()
                         except IOError as e:
                             if e.errno == errno.ESPIPE:
                                 self._seekable = False
                             else:
                                 raise
                     return None
             # These are only the static capabilities.
             # Check the 'getrepocaps' function for the rest.
             capabilities = {'HG20': (),
                             'bookmarks': (),
                             'error': ('abort', 'unsupportedcontent', 'pushraced',
                                       'pushkey'),
                             'listkeys': (),
                             'pushkey': (),
                             'digests': tuple(sorted(util.DIGESTS.keys())),
                             'remote-changegroup': ('http', 'https'),
                             'hgtagsfnodes': (),
                             'rev-branch-cache': (),
                             'phases': ('heads',),
                             'stream': ('v2',),
                            }
             def getrepocaps(repo, allowpushback=False, role=None):
                 """return the bundle2 capabilities for a given repo
                 Exists to allow extensions (like evolution) to mutate the capabilities.
                 The returned value is used for servers advertising their capabilities as
                 well as clients advertising their capabilities to servers as part of
                 bundle2 requests. The ``role`` argument specifies which is which.
                 """
                 if role not in ('client', 'server'):
                     raise error.ProgrammingError('role argument must be client or server')
                 caps = capabilities.copy()
                 caps['changegroup'] = tuple(sorted(
                     changegroup.supportedincomingversions(repo)))
                 if obsolete.isenabled(repo, obsolete.exchangeopt):
                     supportedformat = tuple('V%i' % v for v in obsolete.formats)
                     caps['obsmarkers'] = supportedformat
                 if allowpushback:
                     caps['pushback'] = ()
                 cpmode = repo.ui.config('server', 'concurrent-push-mode')
                 if cpmode == 'check-related':
                     caps['checkheads'] = ('related',)
                 if 'phases' in repo.ui.configlist('devel', 'legacy.exchange'):
                     caps.pop('phases')
                 # Don't advertise stream clone support in server mode if not configured.
                 if role == 'server':
                     streamsupported = repo.ui.configbool('server', 'uncompressed',
                                                          untrusted=True)
                     featuresupported = repo.ui.configbool('experimental', 'bundle2.stream')
                     if not streamsupported or not featuresupported:
                         caps.pop('stream')
                 # Else always advertise support on client, because payload support
                 # should always be advertised.
                 return caps
             def bundle2caps(remote):
                 """return the bundle capabilities of a peer as dict"""
                 raw = remote.capable('bundle2')
                 if not raw and raw != '':
                     return {}
                 capsblob = urlreq.unquote(remote.capable('bundle2'))
                 return decodecaps(capsblob)
             def obsmarkersversion(caps):
                 """extract the list of supported obsmarkers versions from a bundle2caps dict
                 """
                 obscaps = caps.get('obsmarkers', ())
                 return [int(c[1:]) for c in obscaps if c.startswith('V')]
             def writenewbundle(ui, repo, source, filename, bundletype, outgoing, opts,
                                vfs=None, compression=None, compopts=None):
                 if bundletype.startswith('HG10'):
                     cg = changegroup.makechangegroup(repo, outgoing, '01', source)
                     return writebundle(ui, cg, filename, bundletype, vfs=vfs,
                                        compression=compression, compopts=compopts)
                 elif not bundletype.startswith('HG20'):
                     raise error.ProgrammingError('unknown bundle type: %s' % bundletype)
                 caps = {}
                 if 'obsolescence' in opts:
                     caps['obsmarkers'] = ('V1',)
                 bundle = bundle20(ui, caps)
                 bundle.setcompression(compression, compopts)
                 _addpartsfromopts(ui, repo, bundle, source, outgoing, opts)
                 chunkiter = bundle.getchunks()
                 return changegroup.writechunks(ui, chunkiter, filename, vfs=vfs)
             def _addpartsfromopts(ui, repo, bundler, source, outgoing, opts):
                 # We should eventually reconcile this logic with the one behind
                 # 'exchange.getbundle2partsgenerator'.
                 #
                 # The type of input from 'getbundle' and 'writenewbundle' are a bit
                 # different right now. So we keep them separated for now for the sake of
                 # simplicity.
                 # we might not always want a changegroup in such bundle, for example in
                 # stream bundles
                 if opts.get('changegroup', True):
                     cgversion = opts.get('cg.version')
                     if cgversion is None:
                         cgversion = changegroup.safeversion(repo)
                     cg = changegroup.makechangegroup(repo, outgoing, cgversion, source)
                     part = bundler.newpart('changegroup', data=cg.getchunks())
                     part.addparam('version', cg.version)
                     if 'clcount' in cg.extras:
                         part.addparam('nbchanges', '%d' % cg.extras['clcount'],
                                       mandatory=False)
                     if opts.get('phases') and repo.revs('%ln and secret()',
                                                         outgoing.missingheads):
                         part.addparam('targetphase', '%d' % phases.secret, mandatory=False)
                 if opts.get('streamv2', False):
                     addpartbundlestream2(bundler, repo, stream=True)
                 if opts.get('tagsfnodescache', True):
                     addparttagsfnodescache(repo, bundler, outgoing)
                 if opts.get('revbranchcache', True):
                     addpartrevbranchcache(repo, bundler, outgoing)
                 if opts.get('obsolescence', False):
                     obsmarkers = repo.obsstore.relevantmarkers(outgoing.missing)
                     buildobsmarkerspart(bundler, obsmarkers)
                 if opts.get('phases', False):
                     headsbyphase = phases.subsetphaseheads(repo, outgoing.missing)
                     phasedata = phases.binaryencode(headsbyphase)
                     bundler.newpart('phase-heads', data=phasedata)
             def addparttagsfnodescache(repo, bundler, outgoing):
                 # we include the tags fnode cache for the bundle changeset
                 # (as an optional parts)
                 cache = tags.hgtagsfnodescache(repo.unfiltered())
                 chunks = []
                 # .hgtags fnodes are only relevant for head changesets. While we could
                 # transfer values for all known nodes, there will likely be little to
                 # no benefit.
                 #
                 # We don't bother using a generator to produce output data because
                 # a) we only have 40 bytes per head and even esoteric numbers of heads
                 # consume little memory (1M heads is 40MB) b) we don't want to send the
                 # part if we don't have entries and knowing if we have entries requires
                 # cache lookups.
                 for node in outgoing.missingheads:
                     # Don't compute missing, as this may slow down serving.
                     fnode = cache.getfnode(node, computemissing=False)
                     if fnode is not None:
                         chunks.extend([node, fnode])
                 if chunks:
                     bundler.newpart('hgtagsfnodes', data=''.join(chunks))
             def addpartrevbranchcache(repo, bundler, outgoing):
                 # we include the rev branch cache for the bundle changeset
                 # (as an optional parts)
                 cache = repo.revbranchcache()
                 cl = repo.unfiltered().changelog
                 branchesdata = collections.defaultdict(lambda: (set(), set()))
                 for node in outgoing.missing:
                     branch, close = cache.branchinfo(cl.rev(node))
                     branchesdata[branch][close].add(node)
                 def generate():
                     for branch, (nodes, closed) in sorted(branchesdata.items()):
                         utf8branch = encoding.fromlocal(branch)
                         yield rbcstruct.pack(len(utf8branch), len(nodes), len(closed))
                         yield utf8branch
                         for n in sorted(nodes):
                             yield n
                         for n in sorted(closed):
                             yield n
                 bundler.newpart('cache:rev-branch-cache', data=generate(),
                                 mandatory=False)
             def _formatrequirementsspec(requirements):
                 return urlreq.quote(','.join(sorted(requirements)))
             def _formatrequirementsparams(requirements):
                 requirements = _formatrequirementsspec(requirements)
                 params = "%s%s" % (urlreq.quote("requirements="), requirements)
                 return params
             def addpartbundlestream2(bundler, repo, **kwargs):
                 if not kwargs.get('stream', False):
                     return
                 if not streamclone.allowservergeneration(repo):
                     raise error.Abort(_('stream data requested but server does not allow '
                                         'this feature'),
                                       hint=_('well-behaved clients should not be '
                                              'requesting stream data from servers not '
                                              'advertising it; the client may be buggy'))
                 # Stream clones don't compress well. And compression undermines a
                 # goal of stream clones, which is to be fast. Communicate the desire
                 # to avoid compression to consumers of the bundle.
                 bundler.prefercompressed = False
                 filecount, bytecount, it = streamclone.generatev2(repo)
                 requirements = _formatrequirementsspec(repo.requirements)
                 part = bundler.newpart('stream2', data=it)
                 part.addparam('bytecount', '%d' % bytecount, mandatory=True)
                 part.addparam('filecount', '%d' % filecount, mandatory=True)
                 part.addparam('requirements', requirements, mandatory=True)
             def buildobsmarkerspart(bundler, markers):
                 """add an obsmarker part to the bundler with <markers>
                 No part is created if markers is empty.
                 Raises ValueError if the bundler doesn't support any known obsmarker format.
                 """
                 if not markers:
                     return None
                 remoteversions = obsmarkersversion(bundler.capabilities)
                 version = obsolete.commonversion(remoteversions)
                 if version is None:
                     raise ValueError('bundler does not support common obsmarker format')
                 stream = obsolete.encodemarkers(markers, True, version=version)
                 return bundler.newpart('obsmarkers', data=stream)
             def writebundle(ui, cg, filename, bundletype, vfs=None, compression=None,
                             compopts=None):
                 """Write a bundle file and return its filename.
                 Existing files will not be overwritten.
                 If no filename is specified, a temporary file is created.
                 bz2 compression can be turned off.
                 The bundle file will be deleted in case of errors.
                 """
                 if bundletype == "HG20":
                     bundle = bundle20(ui)
                     bundle.setcompression(compression, compopts)
                     part = bundle.newpart('changegroup', data=cg.getchunks())
                     part.addparam('version', cg.version)
                     if 'clcount' in cg.extras:
                         part.addparam('nbchanges', '%d' % cg.extras['clcount'],
                                       mandatory=False)
                     chunkiter = bundle.getchunks()
                 else:
                     # compression argument is only for the bundle2 case
                     assert compression is None
                     if cg.version != '01':
                         raise error.Abort(_('old bundle types only supports v1 '
                                             'changegroups'))
                     header, comp = bundletypes[bundletype]
                     if comp not in util.compengines.supportedbundletypes:
                         raise error.Abort(_('unknown stream compression type: %s')
                                           % comp)
                     compengine = util.compengines.forbundletype(comp)
                     def chunkiter():
                         yield header
                         for chunk in compengine.compressstream(cg.getchunks(), compopts):
                             yield chunk
                     chunkiter = chunkiter()
                 # parse the changegroup data, otherwise we will block
                 # in case of sshrepo because we don't know the end of the stream
                 return changegroup.writechunks(ui, chunkiter, filename, vfs=vfs)
             def combinechangegroupresults(op):
                 """logic to combine 0 or more addchangegroup results into one"""
                 results = [r.get('return', 0)
                            for r in op.records['changegroup']]
                 changedheads = 0
                 result = 1
                 for ret in results:
                     # If any changegroup result is 0, return 0
                     if ret == 0:
                         result = 0
                         break
                     if ret < -1:
                         changedheads += ret + 1
                     elif ret > 1:
                         changedheads += ret - 1
                 if changedheads > 0:
                     result = 1 + changedheads
                 elif changedheads < 0:
                     result = -1 + changedheads
                 return result
             @parthandler('changegroup', ('version', 'nbchanges', 'treemanifest',
                                          'targetphase'))
             def handlechangegroup(op, inpart):
                 """apply a changegroup part on the repo
                 This is a very early implementation that will massive rework before being
                 inflicted to any end-user.
                 """
                 tr = op.gettransaction()
                 unpackerversion = inpart.params.get('version', '01')
                 # We should raise an appropriate exception here
                 cg = changegroup.getunbundler(unpackerversion, inpart, None)
                 # the source and url passed here are overwritten by the one contained in
                 # the transaction.hookargs argument. So 'bundle2' is a placeholder
                 nbchangesets = None
                 if 'nbchanges' in inpart.params:
                     nbchangesets = int(inpart.params.get('nbchanges'))
                 if ('treemanifest' in inpart.params and
                     'treemanifest' not in op.repo.requirements):
                     if len(op.repo.changelog) != 0:
                         raise error.Abort(_(
                             "bundle contains tree manifests, but local repo is "
                             "non-empty and does not use tree manifests"))
                     op.repo.requirements.add('treemanifest')
                     op.repo._applyopenerreqs()
                     op.repo._writerequirements()
                 extrakwargs = {}
                 targetphase = inpart.params.get('targetphase')
                 if targetphase is not None:
                     extrakwargs[r'targetphase'] = int(targetphase)
                 ret = _processchangegroup(op, cg, tr, 'bundle2', 'bundle2',
                                           expectedtotal=nbchangesets, **extrakwargs)
                 if op.reply is not None:
                     # This is definitely not the final form of this
                     # return. But one need to start somewhere.
                     part = op.reply.newpart('reply:changegroup', mandatory=False)
                     part.addparam(
                         'in-reply-to', pycompat.bytestr(inpart.id), mandatory=False)
                     part.addparam('return', '%i' % ret, mandatory=False)
                 assert not inpart.read()
             _remotechangegroupparams = tuple(['url', 'size', 'digests'] +
                 ['digest:%s' % k for k in util.DIGESTS.keys()])
             @parthandler('remote-changegroup', _remotechangegroupparams)
             def handleremotechangegroup(op, inpart):
                 """apply a bundle10 on the repo, given an url and validation information
                 All the information about the remote bundle to import are given as
                 parameters. The parameters include:
                   - url: the url to the bundle10.
                   - size: the bundle10 file size. It is used to validate what was
                     retrieved by the client matches the server knowledge about the bundle.
                   - digests: a space separated list of the digest types provided as
                     parameters.
                   - digest:<digest-type>: the hexadecimal representation of the digest with
                     that name. Like the size, it is used to validate what was retrieved by
                     the client matches what the server knows about the bundle.
                 When multiple digest types are given, all of them are checked.
                 """
                 try:
                     raw_url = inpart.params['url']
                 except KeyError:
                     raise error.Abort(_('remote-changegroup: missing "%s" param') % 'url')
                 parsed_url = util.url(raw_url)
                 if parsed_url.scheme not in capabilities['remote-changegroup']:
                     raise error.Abort(_('remote-changegroup does not support %s urls') %
                         parsed_url.scheme)
                 try:
                     size = int(inpart.params['size'])
                 except ValueError:
                     raise error.Abort(_('remote-changegroup: invalid value for param "%s"')
                         % 'size')
                 except KeyError:
                     raise error.Abort(_('remote-changegroup: missing "%s" param') % 'size')
                 digests = {}
                 for typ in inpart.params.get('digests', '').split():
                     param = 'digest:%s' % typ
                     try:
                         value = inpart.params[param]
                     except KeyError:
                         raise error.Abort(_('remote-changegroup: missing "%s" param') %
                             param)
                     digests[typ] = value
                 real_part = util.digestchecker(url.open(op.ui, raw_url), size, digests)
                 tr = op.gettransaction()
                 from . import exchange
                 cg = exchange.readbundle(op.repo.ui, real_part, raw_url)
                 if not isinstance(cg, changegroup.cg1unpacker):
                     raise error.Abort(_('%s: not a bundle version 1.0') %
                         util.hidepassword(raw_url))
                 ret = _processchangegroup(op, cg, tr, 'bundle2', 'bundle2')
                 if op.reply is not None:
                     # This is definitely not the final form of this
                     # return. But one need to start somewhere.
                     part = op.reply.newpart('reply:changegroup')
                     part.addparam(
                         'in-reply-to', pycompat.bytestr(inpart.id), mandatory=False)
                     part.addparam('return', '%i' % ret, mandatory=False)
                 try:
                     real_part.validate()
                 except error.Abort as e:
                     raise error.Abort(_('bundle at %s is corrupted:\n%s') %
-                        (util.hidepassword(raw_url), str(e)))
+                                      (util.hidepassword(raw_url), bytes(e)))
                 assert not inpart.read()
             @parthandler('reply:changegroup', ('return', 'in-reply-to'))
             def handlereplychangegroup(op, inpart):
                 ret = int(inpart.params['return'])
                 replyto = int(inpart.params['in-reply-to'])
                 op.records.add('changegroup', {'return': ret}, replyto)
             @parthandler('check:bookmarks')
             def handlecheckbookmarks(op, inpart):
                 """check location of bookmarks
                 This part is to be used to detect push race regarding bookmark, it
                 contains binary encoded (bookmark, node) tuple. If the local state does
                 not marks the one in the part, a PushRaced exception is raised
                 """
                 bookdata = bookmarks.binarydecode(inpart)
                 msgstandard = ('repository changed while pushing - please try again '
                                '(bookmark "%s" move from %s to %s)')
                 msgmissing = ('repository changed while pushing - please try again '
                               '(bookmark "%s" is missing, expected %s)')
                 msgexist = ('repository changed while pushing - please try again '
                             '(bookmark "%s" set on %s, expected missing)')
                 for book, node in bookdata:
                     currentnode = op.repo._bookmarks.get(book)
                     if currentnode != node:
                         if node is None:
                             finalmsg = msgexist % (book, nodemod.short(currentnode))
                         elif currentnode is None:
                             finalmsg = msgmissing % (book, nodemod.short(node))
                         else:
                             finalmsg = msgstandard % (book, nodemod.short(node),
                                                       nodemod.short(currentnode))
                         raise error.PushRaced(finalmsg)
             @parthandler('check:heads')
             def handlecheckheads(op, inpart):
                 """check that head of the repo did not change
                 This is used to detect a push race when using unbundle.
                 This replaces the "heads" argument of unbundle."""
                 h = inpart.read(20)
                 heads = []
                 while len(h) == 20:
                     heads.append(h)
                     h = inpart.read(20)
                 assert not h
                 # Trigger a transaction so that we are guaranteed to have the lock now.
                 if op.ui.configbool('experimental', 'bundle2lazylocking'):
                     op.gettransaction()
                 if sorted(heads) != sorted(op.repo.heads()):
                     raise error.PushRaced('repository changed while pushing - '
                                           'please try again')
             @parthandler('check:updated-heads')
             def handlecheckupdatedheads(op, inpart):
                 """check for race on the heads touched by a push
                 This is similar to 'check:heads' but focus on the heads actually updated
                 during the push. If other activities happen on unrelated heads, it is
                 ignored.
                 This allow server with high traffic to avoid push contention as long as
                 unrelated parts of the graph are involved."""
                 h = inpart.read(20)
                 heads = []
                 while len(h) == 20:
                     heads.append(h)
                     h = inpart.read(20)
                 assert not h
                 # trigger a transaction so that we are guaranteed to have the lock now.
                 if op.ui.configbool('experimental', 'bundle2lazylocking'):
                     op.gettransaction()
                 currentheads = set()
                 for ls in op.repo.branchmap().itervalues():
                     currentheads.update(ls)
                 for h in heads:
                     if h not in currentheads:
                         raise error.PushRaced('repository changed while pushing - '
                                               'please try again')
             @parthandler('check:phases')
             def handlecheckphases(op, inpart):
                 """check that phase boundaries of the repository did not change
                 This is used to detect a push race.
                 """
                 phasetonodes = phases.binarydecode(inpart)
                 unfi = op.repo.unfiltered()
                 cl = unfi.changelog
                 phasecache = unfi._phasecache
                 msg = ('repository changed while pushing - please try again '
                        '(%s is %s expected %s)')
                 for expectedphase, nodes in enumerate(phasetonodes):
                     for n in nodes:
                         actualphase = phasecache.phase(unfi, cl.rev(n))
                         if actualphase != expectedphase:
                             finalmsg = msg % (nodemod.short(n),
                                               phases.phasenames[actualphase],
                                               phases.phasenames[expectedphase])
                             raise error.PushRaced(finalmsg)
             @parthandler('output')
             def handleoutput(op, inpart):
                 """forward output captured on the server to the client"""
                 for line in inpart.read().splitlines():
                     op.ui.status(_('remote: %s\n') % line)
             @parthandler('replycaps')
             def handlereplycaps(op, inpart):
                 """Notify that a reply bundle should be created
                 The payload contains the capabilities information for the reply"""
                 caps = decodecaps(inpart.read())
                 if op.reply is None:
                     op.reply = bundle20(op.ui, caps)
             class AbortFromPart(error.Abort):
                 """Sub-class of Abort that denotes an error from a bundle2 part."""
             @parthandler('error:abort', ('message', 'hint'))
             def handleerrorabort(op, inpart):
                 """Used to transmit abort error over the wire"""
                 raise AbortFromPart(inpart.params['message'],
                                     hint=inpart.params.get('hint'))
             @parthandler('error:pushkey', ('namespace', 'key', 'new', 'old', 'ret',
                                            'in-reply-to'))
             def handleerrorpushkey(op, inpart):
                 """Used to transmit failure of a mandatory pushkey over the wire"""
                 kwargs = {}
                 for name in ('namespace', 'key', 'new', 'old', 'ret'):
                     value = inpart.params.get(name)
                     if value is not None:
                         kwargs[name] = value
                 raise error.PushkeyFailed(inpart.params['in-reply-to'],
                                           **pycompat.strkwargs(kwargs))
             @parthandler('error:unsupportedcontent', ('parttype', 'params'))
             def handleerrorunsupportedcontent(op, inpart):
                 """Used to transmit unknown content error over the wire"""
                 kwargs = {}
                 parttype = inpart.params.get('parttype')
                 if parttype is not None:
                     kwargs['parttype'] = parttype
                 params = inpart.params.get('params')
                 if params is not None:
                     kwargs['params'] = params.split('\0')
                 raise error.BundleUnknownFeatureError(**pycompat.strkwargs(kwargs))
             @parthandler('error:pushraced', ('message',))
             def handleerrorpushraced(op, inpart):
                 """Used to transmit push race error over the wire"""
                 raise error.ResponseError(_('push failed:'), inpart.params['message'])
             @parthandler('listkeys', ('namespace',))
             def handlelistkeys(op, inpart):
                 """retrieve pushkey namespace content stored in a bundle2"""
                 namespace = inpart.params['namespace']
                 r = pushkey.decodekeys(inpart.read())
                 op.records.add('listkeys', (namespace, r))
             @parthandler('pushkey', ('namespace', 'key', 'old', 'new'))
             def handlepushkey(op, inpart):
                 """process a pushkey request"""
                 dec = pushkey.decode
                 namespace = dec(inpart.params['namespace'])
                 key = dec(inpart.params['key'])
                 old = dec(inpart.params['old'])
                 new = dec(inpart.params['new'])
                 # Grab the transaction to ensure that we have the lock before performing the
                 # pushkey.
                 if op.ui.configbool('experimental', 'bundle2lazylocking'):
                     op.gettransaction()
                 ret = op.repo.pushkey(namespace, key, old, new)
                 record = {'namespace': namespace,
                           'key': key,
                           'old': old,
                           'new': new}
                 op.records.add('pushkey', record)
                 if op.reply is not None:
                     rpart = op.reply.newpart('reply:pushkey')
                     rpart.addparam(
                         'in-reply-to', pycompat.bytestr(inpart.id), mandatory=False)
                     rpart.addparam('return', '%i' % ret, mandatory=False)
                 if inpart.mandatory and not ret:
                     kwargs = {}
                     for key in ('namespace', 'key', 'new', 'old', 'ret'):
                         if key in inpart.params:
                             kwargs[key] = inpart.params[key]
                     raise error.PushkeyFailed(partid='%d' % inpart.id,
                                               **pycompat.strkwargs(kwargs))
             @parthandler('bookmarks')
             def handlebookmark(op, inpart):
                 """transmit bookmark information
                 The part contains binary encoded bookmark information.
                 The exact behavior of this part can be controlled by the 'bookmarks' mode
                 on the bundle operation.
                 When mode is 'apply' (the default) the bookmark information is applied as
                 is to the unbundling repository. Make sure a 'check:bookmarks' part is
                 issued earlier to check for push races in such update. This behavior is
                 suitable for pushing.
                 When mode is 'records', the information is recorded into the 'bookmarks'
                 records of the bundle operation. This behavior is suitable for pulling.
                 """
                 changes = bookmarks.binarydecode(inpart)
                 pushkeycompat = op.repo.ui.configbool('server', 'bookmarks-pushkey-compat')
                 bookmarksmode = op.modes.get('bookmarks', 'apply')
                 if bookmarksmode == 'apply':
                     tr = op.gettransaction()
                     bookstore = op.repo._bookmarks
                     if pushkeycompat:
                         allhooks = []
                         for book, node in changes:
                             hookargs = tr.hookargs.copy()
                             hookargs['pushkeycompat'] = '1'
                             hookargs['namespace'] = 'bookmarks'
                             hookargs['key'] = book
                             hookargs['old'] = nodemod.hex(bookstore.get(book, ''))
                             hookargs['new'] = nodemod.hex(node if node is not None else '')
                             allhooks.append(hookargs)
                         for hookargs in allhooks:
                             op.repo.hook('prepushkey', throw=True,
                                          **pycompat.strkwargs(hookargs))
                     bookstore.applychanges(op.repo, op.gettransaction(), changes)
                     if pushkeycompat:
                         def runhook():
                             for hookargs in allhooks:
                                 op.repo.hook('pushkey', **pycompat.strkwargs(hookargs))
                         op.repo._afterlock(runhook)
                 elif bookmarksmode == 'records':
                     for book, node in changes:
                         record = {'bookmark': book, 'node': node}
                         op.records.add('bookmarks', record)
                 else:
                     raise error.ProgrammingError('unkown bookmark mode: %s' % bookmarksmode)
             @parthandler('phase-heads')
             def handlephases(op, inpart):
                 """apply phases from bundle part to repo"""
                 headsbyphase = phases.binarydecode(inpart)
                 phases.updatephases(op.repo.unfiltered(), op.gettransaction, headsbyphase)
             @parthandler('reply:pushkey', ('return', 'in-reply-to'))
             def handlepushkeyreply(op, inpart):
                 """retrieve the result of a pushkey request"""
                 ret = int(inpart.params['return'])
                 partid = int(inpart.params['in-reply-to'])
                 op.records.add('pushkey', {'return': ret}, partid)
             @parthandler('obsmarkers')
             def handleobsmarker(op, inpart):
                 """add a stream of obsmarkers to the repo"""
                 tr = op.gettransaction()
                 markerdata = inpart.read()
                 if op.ui.config('experimental', 'obsmarkers-exchange-debug'):
                     op.ui.write(('obsmarker-exchange: %i bytes received\n')
                                 % len(markerdata))
                 # The mergemarkers call will crash if marker creation is not enabled.
                 # we want to avoid this if the part is advisory.
                 if not inpart.mandatory and op.repo.obsstore.readonly:
                     op.repo.ui.debug('ignoring obsolescence markers, feature not enabled\n')
                     return
                 new = op.repo.obsstore.mergemarkers(tr, markerdata)
                 op.repo.invalidatevolatilesets()
                 if new:
                     op.repo.ui.status(_('%i new obsolescence markers\n') % new)
                 op.records.add('obsmarkers', {'new': new})
                 if op.reply is not None:
                     rpart = op.reply.newpart('reply:obsmarkers')
                     rpart.addparam(
                         'in-reply-to', pycompat.bytestr(inpart.id), mandatory=False)
                     rpart.addparam('new', '%i' % new, mandatory=False)
             @parthandler('reply:obsmarkers', ('new', 'in-reply-to'))
             def handleobsmarkerreply(op, inpart):
                 """retrieve the result of a pushkey request"""
                 ret = int(inpart.params['new'])
                 partid = int(inpart.params['in-reply-to'])
                 op.records.add('obsmarkers', {'new': ret}, partid)
             @parthandler('hgtagsfnodes')
             def handlehgtagsfnodes(op, inpart):
                 """Applies .hgtags fnodes cache entries to the local repo.
                 Payload is pairs of 20 byte changeset nodes and filenodes.
                 """
                 # Grab the transaction so we ensure that we have the lock at this point.
                 if op.ui.configbool('experimental', 'bundle2lazylocking'):
                     op.gettransaction()
                 cache = tags.hgtagsfnodescache(op.repo.unfiltered())
                 count = 0
                 while True:
                     node = inpart.read(20)
                     fnode = inpart.read(20)
                     if len(node) < 20 or len(fnode) < 20:
                         op.ui.debug('ignoring incomplete received .hgtags fnodes data\n')
                         break
                     cache.setfnode(node, fnode)
                     count += 1
                 cache.write()
                 op.ui.debug('applied %i hgtags fnodes cache entries\n' % count)
             rbcstruct = struct.Struct('>III')
             @parthandler('cache:rev-branch-cache')
             def handlerbc(op, inpart):
                 """receive a rev-branch-cache payload and update the local cache
                 The payload is a series of data related to each branch
 ) branch name length
 ) number of open heads
 ) number of closed heads
 ) open heads nodes
 ) closed heads nodes
                 """
                 total = 0
                 rawheader = inpart.read(rbcstruct.size)
                 cache = op.repo.revbranchcache()
                 cl = op.repo.unfiltered().changelog
                 while rawheader:
                     header = rbcstruct.unpack(rawheader)
                     total += header[1] + header[2]
                     utf8branch = inpart.read(header[0])
                     branch = encoding.tolocal(utf8branch)
                     for x in xrange(header[1]):
                         node = inpart.read(20)
                         rev = cl.rev(node)
                         cache.setdata(branch, rev, node, False)
                     for x in xrange(header[2]):
                         node = inpart.read(20)
                         rev = cl.rev(node)
                         cache.setdata(branch, rev, node, True)
                     rawheader = inpart.read(rbcstruct.size)
                 cache.write()
             @parthandler('pushvars')
             def bundle2getvars(op, part):
                 '''unbundle a bundle2 containing shellvars on the server'''
                 # An option to disable unbundling on server-side for security reasons
                 if op.ui.configbool('push', 'pushvars.server'):
                     hookargs = {}
                     for key, value in part.advisoryparams:
                         key = key.upper()
                         # We want pushed variables to have USERVAR_ prepended so we know
                         # they came from the --pushvar flag.
                         key = "USERVAR_" + key
                         hookargs[key] = value
                     op.addhookargs(hookargs)
             @parthandler('stream2', ('requirements', 'filecount', 'bytecount'))
             def handlestreamv2bundle(op, part):
                 requirements = urlreq.unquote(part.params['requirements']).split(',')
                 filecount = int(part.params['filecount'])
                 bytecount = int(part.params['bytecount'])
                 repo = op.repo
                 if len(repo):
                     msg = _('cannot apply stream clone to non empty repository')
                     raise error.Abort(msg)
                 repo.ui.debug('applying stream bundle\n')
                 streamclone.applybundlev2(repo, part, filecount, bytecount,
                                           requirements)