1
# Copyright (C) 2020 Breezy Developers
3
# This program is free software; you can redistribute it and/or modify
4
# it under the terms of the GNU General Public License as published by
5
# the Free Software Foundation; either version 2 of the License, or
6
# (at your option) any later version.
8
# This program is distributed in the hope that it will be useful,
9
# but WITHOUT ANY WARRANTY; without even the implied warranty of
10
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
11
# GNU General Public License for more details.
13
# You should have received a copy of the GNU General Public License
14
# along with this program; if not, write to the Free Software
15
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
17
"""MemoryBranch object.
20
from __future__ import absolute_import
22
from . import config as _mod_config, errors, osutils
23
from .branch import Branch
24
from .lock import LogicalLockResult, _RelockDebugMixin
25
from .revision import NULL_REVISION
26
from .tag import DisabledTags, MemoryTags
29
class MemoryBranch(Branch, _RelockDebugMixin):
31
def __init__(self, repository, last_revision_info, tags=None):
32
self.repository = repository
33
self._last_revision_info = last_revision_info
34
self._revision_history_cache = None
36
self.tags = MemoryTags(tags)
38
self.tags = DisabledTags(self)
39
self._partial_revision_history_cache = []
40
self._last_revision_info_cache = None
41
self._revision_id_to_revno_cache = None
42
self._partial_revision_id_to_revno_cache = {}
43
self.base = 'memory://' + osutils.rand_chars(10)
46
return _mod_config.Config()
49
self.repository.lock_read()
50
return LogicalLockResult(self.unlock)
52
def lock_write(self, token=None):
53
self.repository.lock_write()
54
return BranchWriteLockResult(self.unlock, None)
57
self.repository.unlock()
59
def last_revision_info(self):
60
return self._last_revision_info
62
def _gen_revision_history(self):
63
"""Generate the revision history from last revision
65
last_revno, last_revision = self.last_revision_info()
66
with self.lock_read():
67
self._extend_partial_history()
68
return list(reversed(self._partial_revision_history_cache))
70
def get_rev_id(self, revno, history=None):
71
"""Find the revision id of the specified revno."""
72
with self.lock_read():
75
last_revno, last_revid = self.last_revision_info()
76
if revno == last_revno:
78
if last_revno is None:
79
self._extend_partial_history()
80
return self._partial_revision_history_cache[
81
len(self._partial_revision_history_cache) - revno]
83
if revno <= 0 or revno > last_revno:
84
raise errors.NoSuchRevision(self, revno)
85
distance_from_last = last_revno - revno
86
if len(self._partial_revision_history_cache) <= \
88
self._extend_partial_history(distance_from_last)
89
return self._partial_revision_history_cache[distance_from_last]