decode_stream.go

package gojay

import (
	"sync"
	"time"
)

// UnmarshalerStream is the interface to implement for a slice, an array or a slice
// to decode a line delimited JSON to.
type UnmarshalerStream interface {
	UnmarshalStream(*StreamDecoder) error
}

// Stream is a struct holding the Stream api
var Stream = stream{}

type stream struct{}

// A StreamDecoder reads and decodes JSON values from an input stream.
//
// It implements conext.Context and provide a channel to notify interruption.
type StreamDecoder struct {
	mux sync.RWMutex
	*Decoder
	done     chan struct{}
	deadline *time.Time
}

// DecodeStream reads the next line delimited JSON-encoded value from the decoder's input (io.Reader) and stores it in the value pointed to by c.
//
// c must implement UnmarshalerStream. Ideally c is a channel. See example for implementation.
//
// See the documentation for Unmarshal for details about the conversion of JSON into a Go value.
func (dec *StreamDecoder) DecodeStream(c UnmarshalerStream) error {
	if dec.isPooled == 1 {
		panic(InvalidUsagePooledDecoderError("Invalid usage of pooled decoder"))
	}
	if dec.r == nil {
		dec.err = NoReaderError("No reader given to decode stream")
		close(dec.done)
		return dec.err
	}
	for ; dec.cursor < dec.length || dec.read(); dec.cursor++ {
		switch dec.data[dec.cursor] {
		case ' ', '\n', '\t', '\r', ',':
			continue
		default:
			// char is not space start reading
			for dec.nextChar() != 0 {
				// calling unmarshal stream
				err := c.UnmarshalStream(dec)
				if err != nil {
					dec.err = err
					close(dec.done)
					return err
				}
				// garbage collects buffer
				// we don't want the buffer to grow extensively
				dec.data = dec.data[dec.cursor:]
				dec.length = dec.length - dec.cursor
				dec.cursor = 0
			}
			// close the done channel to signal the end of the job
			close(dec.done)
			return nil
		}
	}
	close(dec.done)
	dec.mux.Lock()
	err := dec.raiseInvalidJSONErr(dec.cursor)
	dec.mux.Unlock()
	return err
}

// context.Context implementation

// Done returns a channel that's closed when work is done.
// It implements context.Context
func (dec *StreamDecoder) Done() <-chan struct{} {
	return dec.done
}

// Deadline returns the time when work done on behalf of this context
// should be canceled. Deadline returns ok==false when no deadline is
// set. Successive calls to Deadline return the same results.
func (dec *StreamDecoder) Deadline() (time.Time, bool) {
	if dec.deadline != nil {
		return *dec.deadline, true
	}
	return time.Time{}, false
}

// SetDeadline sets the deadline
func (dec *StreamDecoder) SetDeadline(t time.Time) {
	dec.deadline = &t
}

// Err returns nil if Done is not yet closed.
// If Done is closed, Err returns a non-nil error explaining why.
// It implements context.Context
func (dec *StreamDecoder) Err() error {
	select {
	case <-dec.done:
		dec.mux.RLock()
		defer dec.mux.RUnlock()
		return dec.err
	default:
		return nil
	}
}

// Value implements context.Context
func (dec *StreamDecoder) Value(key interface{}) interface{} {
	return nil
}