msgpack/messagepack.go at main · arivum/msgpack · 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
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
/*
 * Copyright (c) 2022, arivum.
 * All rights reserved.
 * SPDX-License-Identifier: MIT
 * For full license text, see the LICENSE file in the repo root or https://opensource.org/licenses/MIT
 */

package msgpack

import (
	"bufio"
	"io"
	"time"
)

func NewDecoder(r io.Reader) *Decoder {
	var (
		buf = bufio.NewReaderSize(r, 8<<10)
		b   = newPreallocBuf()
		d   = &Decoder{
			underlayingReader: r,
			r:                 buf,
			s:                 make(chan interface{}, 100),
			buf:               b,
		}
	)
	d.mapDecoderFuncsByLen = []func() (map[string]interface{}, error){
		func() (map[string]interface{}, error) {
			return map[string]interface{}{}, nil
		},
		d.decodeMapEnrolledSingleEntry,
		d.decodeMapEnrolledTwoEntries,
		d.decodeMapEnrolledThreeEntries,
	}
	d.sliceDecoderFuncsByLen = []func() ([]interface{}, error){
		func() ([]interface{}, error) {
			return []interface{}{}, nil
		},
		d.decodeSliceEnrolledSingleEntry,
		d.decodeSliceEnrolledTwoEntries,
		d.decodeSliceEnrolledThreeEntries,
	}
	return d
}

func (d *Decoder) Stream() chan interface{} {
	go func() {
		for {
			select {
			case <-d.closeChan:
				d.Close()
			default:
				var v interface{}
				if d.lastError = d.handleNext(&v); d.lastError != nil {
					if v != nil {
						d.s <- v
					}
					d.Close()
					return
				}
				d.s <- v
			}
		}
	}()

	return d.s
}

func (d *Decoder) Close() {
	d.closeOnce.Do(func() {
		for len(d.s) > 0 {
			time.Sleep(10 * time.Microsecond)
		}
		close(d.s)
	})
}

func (d *Decoder) LastError() error {
	if d.lastError == io.EOF {
		d.lastError = nil
	}
	return d.lastError
}

func (d *Decoder) handleNext(v *interface{}) error {
	var err error

	if d.nextByte, err = d.r.ReadByte(); err != nil {
		return err
	}

	if (d.nextByte & 0xE0) == msgPackFlagFixStr {
		*v, err = d.directDecodeFixedString()
		return err
	}

	switch d.nextByte & 0xF0 {
	case msgPackFlagFixMap:
		*v, err = d.decodeFixMap()
		return err
	case msgPackFlagFixArray:
		*v, err = d.decodeFixSlice()
		return err
	}

	switch d.nextByte {
	case msgPackFlagMap16:
		*v, err = d.decodeMap16()
		return err
	case msgPackFlagMap32:
		*v, err = d.decodeMap32()
		return err
	case msgPackFlagArray16:
		*v, err = d.decodeSlice16()
		return err
	case msgPackFlagArray32:
		*v, err = d.decodeSlice32()
		return err
	case msgPackFlagStr8:
		*v, err = d.directDecodeString8()
		return err
	case msgPackFlagStr16:
		*v, err = d.directDecodeString16()
		return err
	case msgPackFlagStr32:
		*v, err = d.directDecodeString32()
		return err
	case msgPackFlagNil:
		*v = nil
		return nil
	case msgPackFlagBoolFalse:
		*v = false
		return nil
	case msgPackFlagBoolTrue:
		*v = true
		return nil
	default:
		*v, err = d.decodeSimple()
		return err
	}
}