108
109
self._client = client._SmartClient(medium)
110
111
self._client = _client
118
return '%s(%r)' % (self.__class__.__name__, self._client)
120
def _probe_bzrdir(self):
121
medium = self._client._medium
113
122
path = self._path_for_remote_call(self._client)
123
if medium._is_remote_before((2, 1)):
127
self._rpc_open_2_1(path)
129
except errors.UnknownSmartMethod:
130
medium._remember_remote_is_before((2, 1))
133
def _rpc_open_2_1(self, path):
134
response = self._call('BzrDir.open_2.1', path)
135
if response == ('no',):
136
raise errors.NotBranchError(path=self.root_transport.base)
137
elif response[0] == 'yes':
138
if response[1] == 'yes':
139
self._has_working_tree = True
140
elif response[1] == 'no':
141
self._has_working_tree = False
143
raise errors.UnexpectedSmartServerResponse(response)
145
raise errors.UnexpectedSmartServerResponse(response)
147
def _rpc_open(self, path):
114
148
response = self._call('BzrDir.open', path)
115
149
if response not in [('yes',), ('no',)]:
116
150
raise errors.UnexpectedSmartServerResponse(response)
117
151
if response == ('no',):
118
raise errors.NotBranchError(path=transport.base)
152
raise errors.NotBranchError(path=self.root_transport.base)
120
154
def _ensure_real(self):
121
155
"""Ensure that there is a _real_bzrdir set.
249
287
def _get_branch_reference(self):
250
288
path = self._path_for_remote_call(self._client)
251
289
medium = self._client._medium
252
if not medium._is_remote_before((1, 13)):
291
('BzrDir.open_branchV3', (2, 1)),
292
('BzrDir.open_branchV2', (1, 13)),
293
('BzrDir.open_branch', None),
295
for verb, required_version in candidate_calls:
296
if required_version and medium._is_remote_before(required_version):
254
response = self._call('BzrDir.open_branchV2', path)
255
if response[0] not in ('ref', 'branch'):
256
raise errors.UnexpectedSmartServerResponse(response)
299
response = self._call(verb, path)
258
300
except errors.UnknownSmartMethod:
259
medium._remember_remote_is_before((1, 13))
260
response = self._call('BzrDir.open_branch', path)
261
if response[0] != 'ok':
301
if required_version is None:
303
medium._remember_remote_is_before(required_version)
306
if verb == 'BzrDir.open_branch':
307
if response[0] != 'ok':
308
raise errors.UnexpectedSmartServerResponse(response)
309
if response[1] != '':
310
return ('ref', response[1])
312
return ('branch', '')
313
if response[0] not in ('ref', 'branch'):
262
314
raise errors.UnexpectedSmartServerResponse(response)
263
if response[1] != '':
264
return ('ref', response[1])
266
return ('branch', '')
268
317
def _get_tree_branch(self):
269
318
"""See BzrDir._get_tree_branch()."""
549
615
return self._custom_format._fetch_reconcile
551
617
def get_format_description(self):
552
return 'bzr remote repository'
619
return 'Remote: ' + self._custom_format.get_format_description()
554
621
def __eq__(self, other):
555
622
return self.__class__ is other.__class__
557
def check_conversion_target(self, target_format):
558
if self.rich_root_data and not target_format.rich_root_data:
559
raise errors.BadConversionTarget(
560
'Does not support rich root data.', target_format)
561
if (self.supports_tree_reference and
562
not getattr(target_format, 'supports_tree_reference', False)):
563
raise errors.BadConversionTarget(
564
'Does not support nested trees', target_format)
566
624
def network_name(self):
567
625
if self._network_name:
568
626
return self._network_name
1680
1748
def insert_stream(self, stream, src_format, resume_tokens):
1681
1749
target = self.target_repo
1682
1750
target._unstacked_provider.missing_keys.clear()
1751
candidate_calls = [('Repository.insert_stream_1.19', (1, 19))]
1683
1752
if target._lock_token:
1684
verb = 'Repository.insert_stream_locked'
1685
extra_args = (target._lock_token or '',)
1686
required_version = (1, 14)
1753
candidate_calls.append(('Repository.insert_stream_locked', (1, 14)))
1754
lock_args = (target._lock_token or '',)
1688
verb = 'Repository.insert_stream'
1690
required_version = (1, 13)
1756
candidate_calls.append(('Repository.insert_stream', (1, 13)))
1691
1758
client = target._client
1692
1759
medium = client._medium
1693
if medium._is_remote_before(required_version):
1694
# No possible way this can work.
1695
return self._insert_real(stream, src_format, resume_tokens)
1696
1760
path = target.bzrdir._path_for_remote_call(client)
1697
if not resume_tokens:
1698
# XXX: Ugly but important for correctness, *will* be fixed during
1699
# 1.13 cycle. Pushing a stream that is interrupted results in a
1700
# fallback to the _real_repositories sink *with a partial stream*.
1701
# Thats bad because we insert less data than bzr expected. To avoid
1702
# this we do a trial push to make sure the verb is accessible, and
1703
# do not fallback when actually pushing the stream. A cleanup patch
1704
# is going to look at rewinding/restarting the stream/partial
1761
# Probe for the verb to use with an empty stream before sending the
1762
# real stream to it. We do this both to avoid the risk of sending a
1763
# large request that is then rejected, and because we don't want to
1764
# implement a way to buffer, rewind, or restart the stream.
1766
for verb, required_version in candidate_calls:
1767
if medium._is_remote_before(required_version):
1770
# We've already done the probing (and set _is_remote_before) on
1771
# a previous insert.
1706
1774
byte_stream = smart_repo._stream_to_byte_stream([], src_format)
1708
1776
response = client.call_with_body_stream(
1709
(verb, path, '') + extra_args, byte_stream)
1777
(verb, path, '') + lock_args, byte_stream)
1710
1778
except errors.UnknownSmartMethod:
1711
1779
medium._remember_remote_is_before(required_version)
1712
return self._insert_real(stream, src_format, resume_tokens)
1785
return self._insert_real(stream, src_format, resume_tokens)
1786
self._last_inv_record = None
1787
self._last_substream = None
1788
if required_version < (1, 19):
1789
# Remote side doesn't support inventory deltas. Wrap the stream to
1790
# make sure we don't send any. If the stream contains inventory
1791
# deltas we'll interrupt the smart insert_stream request and
1793
stream = self._stop_stream_if_inventory_delta(stream)
1713
1794
byte_stream = smart_repo._stream_to_byte_stream(
1714
1795
stream, src_format)
1715
1796
resume_tokens = ' '.join(resume_tokens)
1716
1797
response = client.call_with_body_stream(
1717
(verb, path, resume_tokens) + extra_args, byte_stream)
1798
(verb, path, resume_tokens) + lock_args, byte_stream)
1718
1799
if response[0][0] not in ('ok', 'missing-basis'):
1719
1800
raise errors.UnexpectedSmartServerResponse(response)
1801
if self._last_substream is not None:
1802
# The stream included an inventory-delta record, but the remote
1803
# side isn't new enough to support them. So we need to send the
1804
# rest of the stream via VFS.
1805
self.target_repo.refresh_data()
1806
return self._resume_stream_with_vfs(response, src_format)
1720
1807
if response[0][0] == 'missing-basis':
1721
1808
tokens, missing_keys = bencode.bdecode_as_tuple(response[0][1])
1722
1809
resume_tokens = tokens
1725
1812
self.target_repo.refresh_data()
1726
1813
return [], set()
1815
def _resume_stream_with_vfs(self, response, src_format):
1816
"""Resume sending a stream via VFS, first resending the record and
1817
substream that couldn't be sent via an insert_stream verb.
1819
if response[0][0] == 'missing-basis':
1820
tokens, missing_keys = bencode.bdecode_as_tuple(response[0][1])
1821
# Ignore missing_keys, we haven't finished inserting yet
1824
def resume_substream():
1825
# Yield the substream that was interrupted.
1826
for record in self._last_substream:
1828
self._last_substream = None
1829
def resume_stream():
1830
# Finish sending the interrupted substream
1831
yield ('inventory-deltas', resume_substream())
1832
# Then simply continue sending the rest of the stream.
1833
for substream_kind, substream in self._last_stream:
1834
yield substream_kind, substream
1835
return self._insert_real(resume_stream(), src_format, tokens)
1837
def _stop_stream_if_inventory_delta(self, stream):
1838
"""Normally this just lets the original stream pass-through unchanged.
1840
However if any 'inventory-deltas' substream occurs it will stop
1841
streaming, and store the interrupted substream and stream in
1842
self._last_substream and self._last_stream so that the stream can be
1843
resumed by _resume_stream_with_vfs.
1846
stream_iter = iter(stream)
1847
for substream_kind, substream in stream_iter:
1848
if substream_kind == 'inventory-deltas':
1849
self._last_substream = substream
1850
self._last_stream = stream_iter
1853
yield substream_kind, substream
1729
1856
class RemoteStreamSource(repository.StreamSource):
1730
1857
"""Stream data from a remote server."""
1733
1860
if (self.from_repository._fallback_repositories and
1734
1861
self.to_format._fetch_order == 'topological'):
1735
1862
return self._real_stream(self.from_repository, search)
1736
return self.missing_parents_chain(search, [self.from_repository] +
1737
self.from_repository._fallback_repositories)
1865
repos = [self.from_repository]
1871
repos.extend(repo._fallback_repositories)
1872
sources.append(repo)
1873
return self.missing_parents_chain(search, sources)
1875
def get_stream_for_missing_keys(self, missing_keys):
1876
self.from_repository._ensure_real()
1877
real_repo = self.from_repository._real_repository
1878
real_source = real_repo._get_source(self.to_format)
1879
return real_source.get_stream_for_missing_keys(missing_keys)
1739
1881
def _real_stream(self, repo, search):
1740
1882
"""Get a stream for search from repo.
1770
1913
return self._real_stream(repo, search)
1771
1914
client = repo._client
1772
1915
medium = client._medium
1773
if medium._is_remote_before((1, 13)):
1774
# streaming was added in 1.13
1775
return self._real_stream(repo, search)
1776
1916
path = repo.bzrdir._path_for_remote_call(client)
1778
search_bytes = repo._serialise_search_result(search)
1779
response = repo._call_with_body_bytes_expecting_body(
1780
'Repository.get_stream',
1781
(path, self.to_format.network_name()), search_bytes)
1782
response_tuple, response_handler = response
1783
except errors.UnknownSmartMethod:
1784
medium._remember_remote_is_before((1,13))
1917
search_bytes = repo._serialise_search_result(search)
1918
args = (path, self.to_format.network_name())
1920
('Repository.get_stream_1.19', (1, 19)),
1921
('Repository.get_stream', (1, 13))]
1923
for verb, version in candidate_verbs:
1924
if medium._is_remote_before(version):
1927
response = repo._call_with_body_bytes_expecting_body(
1928
verb, args, search_bytes)
1929
except errors.UnknownSmartMethod:
1930
medium._remember_remote_is_before(version)
1932
response_tuple, response_handler = response
1785
1936
return self._real_stream(repo, search)
1786
1937
if response_tuple[0] != 'ok':
1787
1938
raise errors.UnexpectedSmartServerResponse(response_tuple)
2668
2830
'Missing key %r in context %r', key_err.args[0], context)
2671
if err.error_verb == 'NoSuchRevision':
2833
if err.error_verb == 'IncompatibleRepositories':
2834
raise errors.IncompatibleRepositories(err.error_args[0],
2835
err.error_args[1], err.error_args[2])
2836
elif err.error_verb == 'NoSuchRevision':
2672
2837
raise NoSuchRevision(find('branch'), err.error_args[0])
2673
2838
elif err.error_verb == 'nosuchrevision':
2674
2839
raise NoSuchRevision(find('repository'), err.error_args[0])
2675
elif err.error_tuple == ('nobranch',):
2676
raise errors.NotBranchError(path=find('bzrdir').root_transport.base)
2840
elif err.error_verb == 'nobranch':
2841
if len(err.error_args) >= 1:
2842
extra = err.error_args[0]
2845
raise errors.NotBranchError(path=find('bzrdir').root_transport.base,
2677
2847
elif err.error_verb == 'norepository':
2678
2848
raise errors.NoRepositoryPresent(find('bzrdir'))
2679
2849
elif err.error_verb == 'LockContention':