-
Notifications
You must be signed in to change notification settings - Fork 63
/
result_serializer.go
121 lines (102 loc) · 2.61 KB
/
result_serializer.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
package reindexer
import (
"fmt"
"github.com/restream/reindexer/v3/bindings"
"github.com/restream/reindexer/v3/cjson"
)
type rawResultItemParams struct {
id int
version int
nsid int
proc int
cptr uintptr
data []byte
}
type rawResultsExtraParam struct {
Tag int
Name string
Data []byte
}
type rawResultQueryParams struct {
flags int
totalcount int
qcount int
count int
aggResults [][]byte
explainResults []byte
}
type resultSerializer struct {
cjson.Serializer
flags int
}
type updatePayloadTypeFunc func(nsid int)
func newSerializer(buf []byte) resultSerializer {
return resultSerializer{
Serializer: cjson.NewSerializer(buf),
}
}
func (s *resultSerializer) readRawtItemParams() (v rawResultItemParams) {
if (s.flags & bindings.ResultsWithItemID) != 0 {
v.id = int(s.GetVarUInt())
v.version = int(s.GetVarUInt())
}
if (s.flags & bindings.ResultsWithNsID) != 0 {
v.nsid = int(s.GetVarUInt())
}
if (s.flags & bindings.ResultsWithPercents) != 0 {
v.proc = int(s.GetVarUInt())
}
switch s.flags & bindings.ResultsFormatMask {
case bindings.ResultsPure:
case bindings.ResultsPtrs:
v.cptr = uintptr(s.GetUInt64())
case bindings.ResultsJson, bindings.ResultsCJson:
v.data = s.GetBytes()
}
return v
}
func (s *resultSerializer) readRawQueryParamsKeepExtras(v *rawResultQueryParams, updatePayloadType ...updatePayloadTypeFunc) {
v.flags = int(s.GetVarUInt())
v.totalcount = int(s.GetVarUInt())
v.qcount = int(s.GetVarUInt())
v.count = int(s.GetVarUInt())
if (v.flags & bindings.ResultsWithPayloadTypes) != 0 {
ptCount := int(s.GetVarUInt())
for i := 0; i < ptCount; i++ {
nsid := int(s.GetVarUInt())
nsname := s.GetVString()
_ = nsname
if (len(updatePayloadType)) != 1 {
panic(fmt.Errorf("Internal error: Got payload types from raw query params, but there are no updatePayloadType"))
}
updatePayloadType[0](nsid)
}
}
s.readExtraResults(v)
s.flags = v.flags
}
func (s *resultSerializer) readRawQueryParams(updatePayloadType ...updatePayloadTypeFunc) (v rawResultQueryParams) {
s.readRawQueryParamsKeepExtras(&v, updatePayloadType...)
return v
}
func (s *resultSerializer) readExtraResults(v *rawResultQueryParams) {
firstAgg := true
for {
tag := s.GetVarUInt()
if tag == bindings.QueryResultEnd {
break
}
if firstAgg {
v.aggResults = v.aggResults[:0]
v.explainResults = v.explainResults[:0]
firstAgg = false
}
data := s.GetBytes()
switch tag {
case bindings.QueryResultExplain:
v.explainResults = data
case bindings.QueryResultAggregation:
v.aggResults = append(v.aggResults, data)
}
}
}