-
Notifications
You must be signed in to change notification settings - Fork 0
/
writer_file_roll_policy.go
139 lines (104 loc) · 2.76 KB
/
writer_file_roll_policy.go
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
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
package ethwal
import (
"context"
"io"
"sync"
"time"
)
type FileRollPolicy interface {
ShouldRoll() bool
Reset()
onWrite(data []byte)
onBlockProcessed(blockNum uint64)
}
type fileSizeRollPolicy struct {
maxSize uint64
bytesWritten uint64
}
func NewFileSizeRollPolicy(maxSize uint64) FileRollPolicy {
return &fileSizeRollPolicy{maxSize: maxSize}
}
func (p *fileSizeRollPolicy) ShouldRoll() bool {
return p.bytesWritten >= p.maxSize
}
func (p *fileSizeRollPolicy) Reset() {
p.bytesWritten = 0
}
func (p *fileSizeRollPolicy) onWrite(data []byte) {
p.bytesWritten += uint64(len(data))
}
func (p *fileSizeRollPolicy) onBlockProcessed(blockNum uint64) {}
// fileStats is a writer that keeps track of the number of bytes written to it.
type writerWrapper struct {
io.Writer
fsrp FileRollPolicy
}
func (w *writerWrapper) Write(p []byte) (n int, err error) {
defer w.fsrp.onWrite(p)
return w.Writer.Write(p)
}
type lastBlockNumberRollPolicy struct {
rollInterval uint64
lastBlockNum uint64
}
func (l *lastBlockNumberRollPolicy) onWrite(data []byte) {}
func NewLastBlockNumberRollPolicy(rollInterval uint64) FileRollPolicy {
return &lastBlockNumberRollPolicy{rollInterval: rollInterval}
}
func (l *lastBlockNumberRollPolicy) ShouldRoll() bool {
return l.lastBlockNum != 0 && l.lastBlockNum%l.rollInterval == 0
}
func (l *lastBlockNumberRollPolicy) Reset() {
// noop
}
func (l *lastBlockNumberRollPolicy) onBlockProcessed(blockNum uint64) {
l.lastBlockNum = blockNum
}
type timeBasedRollPolicy struct {
rollInterval time.Duration
onError func(err error)
rollFunc func() error
bgCtx context.Context
bgCancel context.CancelFunc
lastTimeRolled time.Time
mu sync.Mutex
}
func NewTimeBasedRollPolicy(rollInterval time.Duration, onError func(err error)) FileRollPolicy {
return &timeBasedRollPolicy{rollInterval: rollInterval, lastTimeRolled: time.Now(), onError: onError}
}
func (t *timeBasedRollPolicy) ShouldRoll() bool {
if time.Since(t.lastTimeRolled) >= t.rollInterval {
return true
}
return false
}
func (t *timeBasedRollPolicy) Reset() {
t.lastTimeRolled = time.Now()
}
func (t *timeBasedRollPolicy) onWrite(data []byte) {}
func (t *timeBasedRollPolicy) onBlockProcessed(blockNum uint64) {}
type FileRollPolicies []FileRollPolicy
func (policies FileRollPolicies) ShouldRoll() bool {
for _, p := range policies {
if p.ShouldRoll() {
return true
}
}
return false
}
func (policies FileRollPolicies) Reset() {
for _, p := range policies {
p.Reset()
}
}
func (policies FileRollPolicies) onWrite(data []byte) {
for _, p := range policies {
p.onWrite(data)
}
}
func (policies FileRollPolicies) onBlockProcessed(blockNum uint64) {
for _, p := range policies {
p.onBlockProcessed(blockNum)
}
}
var _ FileRollPolicy = &fileSizeRollPolicy{}