real_decoder.go 8.9 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468
  1. package sarama
  2. import (
  3. "encoding/binary"
  4. "math"
  5. "github.com/rcrowley/go-metrics"
  6. )
  7. var (
  8. errInvalidArrayLength = PacketDecodingError{"invalid array length"}
  9. errInvalidByteSliceLength = PacketDecodingError{"invalid byteslice length"}
  10. errInvalidStringLength = PacketDecodingError{"invalid string length"}
  11. errVarintOverflow = PacketDecodingError{"varint overflow"}
  12. errUVarintOverflow = PacketDecodingError{"uvarint overflow"}
  13. errInvalidBool = PacketDecodingError{"invalid bool"}
  14. )
  15. type realDecoder struct {
  16. raw []byte
  17. off int
  18. stack []pushDecoder
  19. registry metrics.Registry
  20. }
  21. // primitives
  22. func (rd *realDecoder) getInt8() (int8, error) {
  23. if rd.remaining() < 1 {
  24. rd.off = len(rd.raw)
  25. return -1, ErrInsufficientData
  26. }
  27. tmp := int8(rd.raw[rd.off])
  28. rd.off++
  29. return tmp, nil
  30. }
  31. func (rd *realDecoder) getInt16() (int16, error) {
  32. if rd.remaining() < 2 {
  33. rd.off = len(rd.raw)
  34. return -1, ErrInsufficientData
  35. }
  36. tmp := int16(binary.BigEndian.Uint16(rd.raw[rd.off:]))
  37. rd.off += 2
  38. return tmp, nil
  39. }
  40. func (rd *realDecoder) getInt32() (int32, error) {
  41. if rd.remaining() < 4 {
  42. rd.off = len(rd.raw)
  43. return -1, ErrInsufficientData
  44. }
  45. tmp := int32(binary.BigEndian.Uint32(rd.raw[rd.off:]))
  46. rd.off += 4
  47. return tmp, nil
  48. }
  49. func (rd *realDecoder) getInt64() (int64, error) {
  50. if rd.remaining() < 8 {
  51. rd.off = len(rd.raw)
  52. return -1, ErrInsufficientData
  53. }
  54. tmp := int64(binary.BigEndian.Uint64(rd.raw[rd.off:]))
  55. rd.off += 8
  56. return tmp, nil
  57. }
  58. func (rd *realDecoder) getVarint() (int64, error) {
  59. tmp, n := binary.Varint(rd.raw[rd.off:])
  60. if n == 0 {
  61. rd.off = len(rd.raw)
  62. return -1, ErrInsufficientData
  63. }
  64. if n < 0 {
  65. rd.off -= n
  66. return -1, errVarintOverflow
  67. }
  68. rd.off += n
  69. return tmp, nil
  70. }
  71. func (rd *realDecoder) getUVarint() (uint64, error) {
  72. tmp, n := binary.Uvarint(rd.raw[rd.off:])
  73. if n == 0 {
  74. rd.off = len(rd.raw)
  75. return 0, ErrInsufficientData
  76. }
  77. if n < 0 {
  78. rd.off -= n
  79. return 0, errUVarintOverflow
  80. }
  81. rd.off += n
  82. return tmp, nil
  83. }
  84. func (rd *realDecoder) getFloat64() (float64, error) {
  85. if rd.remaining() < 8 {
  86. rd.off = len(rd.raw)
  87. return -1, ErrInsufficientData
  88. }
  89. tmp := math.Float64frombits(binary.BigEndian.Uint64(rd.raw[rd.off:]))
  90. rd.off += 8
  91. return tmp, nil
  92. }
  93. func (rd *realDecoder) getArrayLength() (int, error) {
  94. if rd.remaining() < 4 {
  95. rd.off = len(rd.raw)
  96. return -1, ErrInsufficientData
  97. }
  98. tmp := int(int32(binary.BigEndian.Uint32(rd.raw[rd.off:])))
  99. rd.off += 4
  100. if tmp > rd.remaining() {
  101. rd.off = len(rd.raw)
  102. return -1, ErrInsufficientData
  103. } else if tmp > 2*math.MaxUint16 {
  104. return -1, errInvalidArrayLength
  105. }
  106. return tmp, nil
  107. }
  108. func (rd *realDecoder) getCompactArrayLength() (int, error) {
  109. n, err := rd.getUVarint()
  110. if err != nil {
  111. return 0, err
  112. }
  113. if n == 0 {
  114. return 0, nil
  115. }
  116. return int(n) - 1, nil
  117. }
  118. func (rd *realDecoder) getBool() (bool, error) {
  119. b, err := rd.getInt8()
  120. if err != nil || b == 0 {
  121. return false, err
  122. }
  123. if b != 1 {
  124. return false, errInvalidBool
  125. }
  126. return true, nil
  127. }
  128. func (rd *realDecoder) getEmptyTaggedFieldArray() (int, error) {
  129. tagCount, err := rd.getUVarint()
  130. if err != nil {
  131. return 0, err
  132. }
  133. // skip over any tagged fields without deserializing them
  134. // as we don't currently support doing anything with them
  135. for i := uint64(0); i < tagCount; i++ {
  136. // fetch and ignore tag identifier
  137. _, err := rd.getUVarint()
  138. if err != nil {
  139. return 0, err
  140. }
  141. length, err := rd.getUVarint()
  142. if err != nil {
  143. return 0, err
  144. }
  145. if _, err := rd.getRawBytes(int(length)); err != nil {
  146. return 0, err
  147. }
  148. }
  149. return 0, nil
  150. }
  151. // collections
  152. func (rd *realDecoder) getBytes() ([]byte, error) {
  153. tmp, err := rd.getInt32()
  154. if err != nil {
  155. return nil, err
  156. }
  157. if tmp == -1 {
  158. return nil, nil
  159. }
  160. return rd.getRawBytes(int(tmp))
  161. }
  162. func (rd *realDecoder) getVarintBytes() ([]byte, error) {
  163. tmp, err := rd.getVarint()
  164. if err != nil {
  165. return nil, err
  166. }
  167. if tmp == -1 {
  168. return nil, nil
  169. }
  170. return rd.getRawBytes(int(tmp))
  171. }
  172. func (rd *realDecoder) getCompactBytes() ([]byte, error) {
  173. n, err := rd.getUVarint()
  174. if err != nil {
  175. return nil, err
  176. }
  177. length := int(n - 1)
  178. return rd.getRawBytes(length)
  179. }
  180. func (rd *realDecoder) getStringLength() (int, error) {
  181. length, err := rd.getInt16()
  182. if err != nil {
  183. return 0, err
  184. }
  185. n := int(length)
  186. switch {
  187. case n < -1:
  188. return 0, errInvalidStringLength
  189. case n > rd.remaining():
  190. rd.off = len(rd.raw)
  191. return 0, ErrInsufficientData
  192. }
  193. return n, nil
  194. }
  195. func (rd *realDecoder) getString() (string, error) {
  196. n, err := rd.getStringLength()
  197. if err != nil || n == -1 {
  198. return "", err
  199. }
  200. tmpStr := string(rd.raw[rd.off : rd.off+n])
  201. rd.off += n
  202. return tmpStr, nil
  203. }
  204. func (rd *realDecoder) getNullableString() (*string, error) {
  205. n, err := rd.getStringLength()
  206. if err != nil || n == -1 {
  207. return nil, err
  208. }
  209. tmpStr := string(rd.raw[rd.off : rd.off+n])
  210. rd.off += n
  211. return &tmpStr, err
  212. }
  213. func (rd *realDecoder) getCompactString() (string, error) {
  214. n, err := rd.getUVarint()
  215. if err != nil {
  216. return "", err
  217. }
  218. length := int(n - 1)
  219. if length < 0 {
  220. return "", errInvalidByteSliceLength
  221. }
  222. tmpStr := string(rd.raw[rd.off : rd.off+length])
  223. rd.off += length
  224. return tmpStr, nil
  225. }
  226. func (rd *realDecoder) getCompactNullableString() (*string, error) {
  227. n, err := rd.getUVarint()
  228. if err != nil {
  229. return nil, err
  230. }
  231. length := int(n - 1)
  232. if length < 0 {
  233. return nil, err
  234. }
  235. tmpStr := string(rd.raw[rd.off : rd.off+length])
  236. rd.off += length
  237. return &tmpStr, err
  238. }
  239. func (rd *realDecoder) getCompactInt32Array() ([]int32, error) {
  240. n, err := rd.getUVarint()
  241. if err != nil {
  242. return nil, err
  243. }
  244. if n == 0 {
  245. return nil, nil
  246. }
  247. arrayLength := int(n) - 1
  248. ret := make([]int32, arrayLength)
  249. for i := range ret {
  250. ret[i] = int32(binary.BigEndian.Uint32(rd.raw[rd.off:]))
  251. rd.off += 4
  252. }
  253. return ret, nil
  254. }
  255. func (rd *realDecoder) getInt32Array() ([]int32, error) {
  256. if rd.remaining() < 4 {
  257. rd.off = len(rd.raw)
  258. return nil, ErrInsufficientData
  259. }
  260. n := int(binary.BigEndian.Uint32(rd.raw[rd.off:]))
  261. rd.off += 4
  262. if rd.remaining() < 4*n {
  263. rd.off = len(rd.raw)
  264. return nil, ErrInsufficientData
  265. }
  266. if n == 0 {
  267. return nil, nil
  268. }
  269. if n < 0 {
  270. return nil, errInvalidArrayLength
  271. }
  272. ret := make([]int32, n)
  273. for i := range ret {
  274. ret[i] = int32(binary.BigEndian.Uint32(rd.raw[rd.off:]))
  275. rd.off += 4
  276. }
  277. return ret, nil
  278. }
  279. func (rd *realDecoder) getInt64Array() ([]int64, error) {
  280. if rd.remaining() < 4 {
  281. rd.off = len(rd.raw)
  282. return nil, ErrInsufficientData
  283. }
  284. n := int(binary.BigEndian.Uint32(rd.raw[rd.off:]))
  285. rd.off += 4
  286. if rd.remaining() < 8*n {
  287. rd.off = len(rd.raw)
  288. return nil, ErrInsufficientData
  289. }
  290. if n == 0 {
  291. return nil, nil
  292. }
  293. if n < 0 {
  294. return nil, errInvalidArrayLength
  295. }
  296. ret := make([]int64, n)
  297. for i := range ret {
  298. ret[i] = int64(binary.BigEndian.Uint64(rd.raw[rd.off:]))
  299. rd.off += 8
  300. }
  301. return ret, nil
  302. }
  303. func (rd *realDecoder) getStringArray() ([]string, error) {
  304. if rd.remaining() < 4 {
  305. rd.off = len(rd.raw)
  306. return nil, ErrInsufficientData
  307. }
  308. n := int(binary.BigEndian.Uint32(rd.raw[rd.off:]))
  309. rd.off += 4
  310. if n == 0 {
  311. return nil, nil
  312. }
  313. if n < 0 {
  314. return nil, errInvalidArrayLength
  315. }
  316. ret := make([]string, n)
  317. for i := range ret {
  318. str, err := rd.getString()
  319. if err != nil {
  320. return nil, err
  321. }
  322. ret[i] = str
  323. }
  324. return ret, nil
  325. }
  326. // subsets
  327. func (rd *realDecoder) remaining() int {
  328. return len(rd.raw) - rd.off
  329. }
  330. func (rd *realDecoder) getSubset(length int) (packetDecoder, error) {
  331. buf, err := rd.getRawBytes(length)
  332. if err != nil {
  333. return nil, err
  334. }
  335. return &realDecoder{raw: buf}, nil
  336. }
  337. func (rd *realDecoder) getRawBytes(length int) ([]byte, error) {
  338. if length < 0 {
  339. return nil, errInvalidByteSliceLength
  340. } else if length > rd.remaining() {
  341. rd.off = len(rd.raw)
  342. return nil, ErrInsufficientData
  343. }
  344. start := rd.off
  345. rd.off += length
  346. return rd.raw[start:rd.off], nil
  347. }
  348. func (rd *realDecoder) peek(offset, length int) (packetDecoder, error) {
  349. if rd.remaining() < offset+length {
  350. return nil, ErrInsufficientData
  351. }
  352. off := rd.off + offset
  353. return &realDecoder{raw: rd.raw[off : off+length]}, nil
  354. }
  355. func (rd *realDecoder) peekInt8(offset int) (int8, error) {
  356. const byteLen = 1
  357. if rd.remaining() < offset+byteLen {
  358. return -1, ErrInsufficientData
  359. }
  360. return int8(rd.raw[rd.off+offset]), nil
  361. }
  362. // stacks
  363. func (rd *realDecoder) push(in pushDecoder) error {
  364. in.saveOffset(rd.off)
  365. var reserve int
  366. if dpd, ok := in.(dynamicPushDecoder); ok {
  367. if err := dpd.decode(rd); err != nil {
  368. return err
  369. }
  370. } else {
  371. reserve = in.reserveLength()
  372. if rd.remaining() < reserve {
  373. rd.off = len(rd.raw)
  374. return ErrInsufficientData
  375. }
  376. }
  377. rd.stack = append(rd.stack, in)
  378. rd.off += reserve
  379. return nil
  380. }
  381. func (rd *realDecoder) pop() error {
  382. // this is go's ugly pop pattern (the inverse of append)
  383. in := rd.stack[len(rd.stack)-1]
  384. rd.stack = rd.stack[:len(rd.stack)-1]
  385. return in.check(rd.off, rd.raw)
  386. }
  387. func (rd *realDecoder) metricRegistry() metrics.Registry {
  388. return rd.registry
  389. }