service/vendor/github.com/bytedance/sonic/internal/decoder/stream.go

218 lines
5.5 KiB
Go

/*
* Copyright 2021 ByteDance Inc.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package decoder
import (
`bytes`
`io`
`sync`
`github.com/bytedance/sonic/option`
`github.com/bytedance/sonic/internal/native/types`
)
var (
minLeftBufferShift uint = 1
)
// StreamDecoder is the decoder context object for streaming input.
type StreamDecoder struct {
r io.Reader
buf []byte
scanp int
scanned int64
err error
Decoder
}
var bufPool = sync.Pool{
New: func () interface{} {
return make([]byte, 0, option.DefaultDecoderBufferSize)
},
}
// NewStreamDecoder adapts to encoding/json.NewDecoder API.
//
// NewStreamDecoder returns a new decoder that reads from r.
func NewStreamDecoder(r io.Reader) *StreamDecoder {
return &StreamDecoder{r : r}
}
// Decode decodes input stream into val with corresponding data.
// Redundantly bytes may be read and left in its buffer, and can be used at next call.
// Either io error from underlying io.Reader (except io.EOF)
// or syntax error from data will be recorded and stop subsequently decoding.
func (self *StreamDecoder) Decode(val interface{}) (err error) {
if self.err != nil {
return self.err
}
var buf = self.buf[self.scanp:]
var p = 0
var recycle bool
if cap(buf) == 0 {
buf = bufPool.Get().([]byte)
recycle = true
}
var first = true
var repeat = true
read_more:
for {
l := len(buf)
realloc(&buf)
n, err := self.r.Read(buf[l:cap(buf)])
buf = buf[:l+n]
if err != nil {
repeat = false
if err == io.EOF {
if len(buf) == 0 {
return err
}
break
}
self.err = err
return err
}
if n > 0 || first {
break
}
}
first = false
l := len(buf)
if l > 0 {
self.Decoder.Reset(string(buf))
err = self.Decoder.Decode(val)
if err != nil {
if repeat && self.repeatable(err) {
goto read_more
}
self.err = err
}
p = self.Decoder.Pos()
self.scanned += int64(p)
self.scanp = 0
}
if l > p {
// remain undecoded bytes, so copy them into self.buf
self.buf = append(self.buf[:0], buf[p:]...)
} else {
self.buf = nil
recycle = true
}
if recycle {
buf = buf[:0]
bufPool.Put(buf)
}
return err
}
func (self StreamDecoder) repeatable(err error) bool {
if ee, ok := err.(SyntaxError); ok &&
(ee.Code == types.ERR_EOF || (ee.Code == types.ERR_INVALID_CHAR && self.i >= len(self.s)-1)) {
return true
}
return false
}
// InputOffset returns the input stream byte offset of the current decoder position.
// The offset gives the location of the end of the most recently returned token and the beginning of the next token.
func (self *StreamDecoder) InputOffset() int64 {
return self.scanned + int64(self.scanp)
}
// Buffered returns a reader of the data remaining in the Decoder's buffer.
// The reader is valid until the next call to Decode.
func (self *StreamDecoder) Buffered() io.Reader {
return bytes.NewReader(self.buf[self.scanp:])
}
// More reports whether there is another element in the
// current array or object being parsed.
func (self *StreamDecoder) More() bool {
if self.err != nil {
return false
}
c, err := self.peek()
return err == nil && c != ']' && c != '}'
}
func (self *StreamDecoder) peek() (byte, error) {
var err error
for {
for i := self.scanp; i < len(self.buf); i++ {
c := self.buf[i]
if isSpace(c) {
continue
}
self.scanp = i
return c, nil
}
// buffer has been scanned, now report any error
if err != nil {
if err != io.EOF {
self.err = err
}
return 0, err
}
err = self.refill()
}
}
func isSpace(c byte) bool {
return types.SPACE_MASK & (1 << c) != 0
}
func (self *StreamDecoder) refill() error {
// Make room to read more into the buffer.
// First slide down data already consumed.
if self.scanp > 0 {
self.scanned += int64(self.scanp)
n := copy(self.buf, self.buf[self.scanp:])
self.buf = self.buf[:n]
self.scanp = 0
}
// Grow buffer if not large enough.
realloc(&self.buf)
// Read. Delay error for next iteration (after scan).
n, err := self.r.Read(self.buf[len(self.buf):cap(self.buf)])
self.buf = self.buf[0 : len(self.buf)+n]
return err
}
func realloc(buf *[]byte) {
l := uint(len(*buf))
c := uint(cap(*buf))
if c - l <= c >> minLeftBufferShift {
e := l+(l>>minLeftBufferShift)
if e < option.DefaultDecoderBufferSize {
e = option.DefaultDecoderBufferSize
}
tmp := make([]byte, l, e)
copy(tmp, *buf)
*buf = tmp
}
}