-
Notifications
You must be signed in to change notification settings - Fork 12
/
rangesend.go
123 lines (96 loc) · 3.16 KB
/
rangesend.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
// Code generated by go generate; DO NOT EDIT.
package inx
import iotago "github.com/iotaledger/iota.go/v4"
type streamRange struct {
start iotago.SlotIndex
end iotago.SlotIndex
lastSent iotago.SlotIndex
}
// tells whether the stream range has a range requested.
func (stream *streamRange) rangeRequested() bool {
return stream.start > 0
}
// tells whether the stream is bounded, aka has an end index.
func (stream *streamRange) isBounded() bool {
return stream.end > 0
}
// handleRangedSend {{- if hasParams}}{{paramCount}}{{end}} handles the sending of data within a streamRange.
// - sendFunc gets executed for the given index.
// - if data wasn't sent between streamRange.lastSent and the given index, then the given catchUpFunc is executed
// with the range from streamRange.lastSent + 1 up to index - 1.
// - streamRange.lastSent is auto. updated
//
func handleRangedSend1[T1 any](index iotago.SlotIndex, arg1 T1, streamRange *streamRange,
catchUpFunc func(start iotago.SlotIndex, end iotago.SlotIndex) error,
sendFunc func(index iotago.SlotIndex, arg1 T1) error,
) (bool, error) {
// below requested range
if streamRange.rangeRequested() && index < streamRange.start {
return false, nil
}
// execute catch up function with missing indices
if streamRange.rangeRequested() && index-1 > streamRange.lastSent {
startIndex := streamRange.start
if startIndex < streamRange.lastSent+1 {
startIndex = streamRange.lastSent + 1
}
endIndex := index - 1
if streamRange.isBounded() && endIndex > streamRange.end {
endIndex = streamRange.end
}
if err := catchUpFunc(startIndex, endIndex); err != nil {
return false, err
}
streamRange.lastSent = endIndex
}
// stream finished
if streamRange.isBounded() && index > streamRange.end {
return true, nil
}
if err := sendFunc(index, arg1); err != nil {
return false, err
}
streamRange.lastSent = index
// stream finished
if streamRange.isBounded() && index >= streamRange.end {
return true, nil
}
return false, nil
}
func handleRangedSend2[T1, T2 any](index iotago.SlotIndex, arg1 T1, arg2 T2, streamRange *streamRange,
catchUpFunc func(start iotago.SlotIndex, end iotago.SlotIndex) error,
sendFunc func(index iotago.SlotIndex, arg1 T1, arg2 T2) error,
) (bool, error) {
// below requested range
if streamRange.rangeRequested() && index < streamRange.start {
return false, nil
}
// execute catch up function with missing indices
if streamRange.rangeRequested() && index-1 > streamRange.lastSent {
startIndex := streamRange.start
if startIndex < streamRange.lastSent+1 {
startIndex = streamRange.lastSent + 1
}
endIndex := index - 1
if streamRange.isBounded() && endIndex > streamRange.end {
endIndex = streamRange.end
}
if err := catchUpFunc(startIndex, endIndex); err != nil {
return false, err
}
streamRange.lastSent = endIndex
}
// stream finished
if streamRange.isBounded() && index > streamRange.end {
return true, nil
}
if err := sendFunc(index, arg1, arg2); err != nil {
return false, err
}
streamRange.lastSent = index
// stream finished
if streamRange.isBounded() && index >= streamRange.end {
return true, nil
}
return false, nil
}