pharo/src/Ombu/OmBlockFileStore.class.st at d18e78e6b7d5dd7d74ade701f30fe29e25ea934d · pharo-project/pharo · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
"
I am a single-file store of entries. Each new entry is appended at the end. Entries are lazily read in blocks (represented by OmBlock) from file on demand.
"
Class {
	#name : 'OmBlockFileStore',
	#superclass : 'OmFileStore',
	#instVars : [
		'blocks'
	],
	#category : 'Ombu-Stores',
	#package : 'Ombu',
	#tag : 'Stores'
}

{ #category : 'private' }
OmBlockFileStore >> blockForLocalNameAsInteger: anIndex [

	self checkIfMustRefresh.
	blocks reverseDo: [:each | each firstLocalNameAsInteger <= anIndex ifTrue: [ ^each ] ].
	^ self error: 'block not found'
]

{ #category : 'accessing' }
OmBlockFileStore >> blockSize [

	^  131072 "2 ** 17"
]

{ #category : 'accessing' }
OmBlockFileStore >> entryPositionsStartingAt: startPosition upTo: endPosition [

	self readEntriesWith: [:readStream |
		readStream position: startPosition.
		^self newEntryReader
			stream: readStream;
			entryPositionsUpTo: endPosition ]
]

{ #category : 'initialization' }
OmBlockFileStore >> initialize [

	super initialize.

	blocks := OrderedCollection with:
		(OmBlock
			store: self
			startPosition: 0
			endPosition: self blockSize - 1
			firstEntryReference: (self referenceToLocalName: '1'))
]

{ #category : 'accessing' }
OmBlockFileStore >> readEntryForLocalName: aString ifPresent: presentBlockClosure ifAbsent: absentBlockClosure [

	^ (self blockForLocalNameAsInteger: aString extractIntegerPart)
		  atLocalName: aString
		  ifPresent: [ :position |
			  presentBlockClosure value: (self nextEntryFromPosition: position) ]
		  ifAbsent: absentBlockClosure
]

{ #category : 'refreshing' }
OmBlockFileStore >> refresh [
	self flush.

	self
		critical: [ mustRefresh := false.
			self initialize.
			self fileReference ifAbsent: [ ^ self ].
			self
				readEntriesWith: [ :readStream |
					self refreshNewBlocksFrom: readStream.
					self refreshHeadReferenceAndEntryCountFrom: readStream ] ]
]

{ #category : 'private' }
OmBlockFileStore >> refreshEntryPositionsByLocalNameStartingAt: firstStreamPosition since: initialLocalName [
	"AKA: Refresh blocks after new entries were written."

	| lastBlock |
	lastBlock := blocks last.

	"Update last block if necessary"
	firstStreamPosition < lastBlock endPosition ifTrue: [
		lastBlock refreshIfNeededStartingAt: firstStreamPosition since: initialLocalName ].

	"Add new blocks if necessary"
	lastStreamPosition > lastBlock endPosition ifTrue: [
		self readEntriesWith: [:readStream | self refreshNewBlocksFrom: readStream ] ]
]

{ #category : 'refreshing' }
OmBlockFileStore >> refreshHeadReferenceAndEntryCountFrom: readStream [

	entryCount := blocks last lastLocalNameAsInteger.
	headReference := self referenceToLocalName: entryCount asString
]

{ #category : 'refreshing' }
OmBlockFileStore >> refreshNewBlocksFrom: readStream [

	| reader blockPosition |
	reader := self newEntryReader.
	reader stream: readStream.

	[
	blockPosition := blocks size * self blockSize.
	blockPosition < readStream size ] whileTrue: [
		| startPosition blockFirstEntry |
		reader positionAtNextEntryStartingAt: blockPosition.
		readStream atEnd ifTrue: [ ^ self "No more entries to add in a new block" ].
		startPosition := readStream position.
		blockFirstEntry := reader nextEntry.
		blocks add: (OmBlock
				 store: self
				 startPosition: startPosition
				 endPosition: blockPosition + self blockSize - 1
				 firstEntryReference: (self referenceTo: blockFirstEntry)) ]
]