decode_stream_test.go

package gojay

import (
	"context"
	"testing"
	"time"

	"github.com/stretchr/testify/assert"
)

// Basic Behaviour Tests
//
func TestDecoderImplementsContext(t *testing.T) {
	var dec interface{} = &StreamDecoder{}
	_ = dec.(context.Context)
}

func TestDecodeStreamNoReader(t *testing.T) {
	dec := Stream.NewDecoder(nil)
	dec.done = make(chan struct{}, 1)
	testChan := ChannelStreamObjects(make(chan *TestObj))
	go dec.DecodeStream(&testChan)

	select {
	case <-dec.Done():
		assert.NotNil(t, dec.Err(), "dec.Err() should not be nil")
		assert.Equal(t, "No reader given to decode stream", dec.Err().Error(), "dec.Err().Error() should not be 'No reader given to decode stream'")
	case <-testChan:
		assert.True(t, false, "should not be called as decoder should not return error right away")
	}
}

//  Table Tests

// Objects

type StreamTestObject struct {
	name         string
	streamReader *StreamReader
	expectations func(error, []*TestObj, *testing.T)
}

func TestStreamDecodingObjectsParallel(t *testing.T) {
	var tests = []StreamTestObject{
		{
			name: "Stream objects",
			streamReader: &StreamReader{
				readChan: make(chan string),
				done:     make(chan struct{}),
				data: `
					{"test":246,"test2":-246,"test3":"string"}
					{"test":247,"test2":248,"test3":"string"}
					{"test":777,"test2":456,"test3":"string"}
					{"test":777,"test2":456,"test3":"string"}
					{"test":777,"test2":456,"test3":"string"}
					{"test":777,"test2":456,"test3":"string"}
				`,
			},
			expectations: func(err error, result []*TestObj, t *testing.T) {
				assert.Nil(t, err, "err should be nil")

				assert.Equal(t, 246, result[0].test, "v[0].test should be equal to 246")
				assert.Equal(t, -246, result[0].test2, "v[0].test2 should be equal to -247")
				assert.Equal(t, "string", result[0].test3, "v[0].test3 should be equal to \"string\"")

				assert.Equal(t, 247, result[1].test, "result[1].test should be equal to 246")
				assert.Equal(t, 248, result[1].test2, "result[1].test2 should be equal to 248")
				assert.Equal(t, "string", result[1].test3, "result[1].test3 should be equal to \"string\"")

				assert.Equal(t, 777, result[2].test, "result[2].test should be equal to 777")
				assert.Equal(t, 456, result[2].test2, "result[2].test2 should be equal to 456")
				assert.Equal(t, "string", result[2].test3, "result[2].test3 should be equal to \"string\"")

				assert.Equal(t, 777, result[3].test, "result[3].test should be equal to 777")
				assert.Equal(t, 456, result[3].test2, "result[3].test2 should be equal to 456")
				assert.Equal(t, "string", result[3].test3, "result[3].test3 should be equal to \"string\"")

				assert.Equal(t, 777, result[4].test, "result[4].test should be equal to 777")
				assert.Equal(t, 456, result[4].test2, "result[4].test2 should be equal to 456")
				assert.Equal(t, "string", result[4].test3, "result[4].test3 should be equal to \"string\"")

				assert.Equal(t, 777, result[5].test, "result[5].test should be equal to 777")
				assert.Equal(t, 456, result[5].test2, "result[5].test2 should be equal to 456")
				assert.Equal(t, "string", result[5].test3, "result[5].test3 should be equal to \"string\"")
			},
		},
		{
			name: "Stream test objects with null values",
			streamReader: &StreamReader{
				readChan: make(chan string),
				done:     make(chan struct{}),
				data: `
					{"test":246,"test2":-246,"test3":"string"}
					{"test":247,"test2":248,"test3":"string"}
					null
					{"test":777,"test2":456,"test3":"string"}
					{"test":777,"test2":456,"test3":"string"}
					{"test":777,"test2":456,"test3":"string"}
				`,
			},
			expectations: func(err error, result []*TestObj, t *testing.T) {
				assert.Nil(t, err, "err should be nil")

				assert.Equal(t, 246, result[0].test, "v[0].test should be equal to 246")
				assert.Equal(t, -246, result[0].test2, "v[0].test2 should be equal to -247")
				assert.Equal(t, "string", result[0].test3, "v[0].test3 should be equal to \"string\"")

				assert.Equal(t, 247, result[1].test, "result[1].test should be equal to 246")
				assert.Equal(t, 248, result[1].test2, "result[1].test2 should be equal to 248")
				assert.Equal(t, "string", result[1].test3, "result[1].test3 should be equal to \"string\"")

				assert.Equal(t, 0, result[2].test, "result[2].test should be equal to 0 as input is null")
				assert.Equal(t, 0, result[2].test2, "result[2].test2 should be equal to 0 as input is null")
				assert.Equal(t, "", result[2].test3, "result[2].test3 should be equal to \"\" as input is null")

				assert.Equal(t, 777, result[3].test, "result[3].test should be equal to 777")
				assert.Equal(t, 456, result[3].test2, "result[3].test2 should be equal to 456")
				assert.Equal(t, "string", result[3].test3, "result[3].test3 should be equal to \"string\"")

				assert.Equal(t, 777, result[4].test, "result[4].test should be equal to 777")
				assert.Equal(t, 456, result[4].test2, "result[4].test2 should be equal to 456")
				assert.Equal(t, "string", result[4].test3, "result[4].test3 should be equal to \"string\"")

				assert.Equal(t, 777, result[5].test, "result[5].test should be equal to 777")
				assert.Equal(t, 456, result[5].test2, "result[5].test2 should be equal to 456")
				assert.Equal(t, "string", result[5].test3, "result[5].test3 should be equal to \"string\"")
			},
		},
		{
			name: "Stream test starting with null values",
			streamReader: &StreamReader{
				readChan: make(chan string),
				done:     make(chan struct{}),
				data: `
					null
					{"test":246,"test2":-246,"test3":"string"}
					{"test":247,"test2":248,"test3":"string"}
				`,
			},
			expectations: func(err error, result []*TestObj, t *testing.T) {
				assert.Nil(t, err, "err should be nil")

				assert.Equal(t, 0, result[0].test, "result[0].test should be equal to 0 as input is null")
				assert.Equal(t, 0, result[0].test2, "result[0].test2 should be equal to 0 as input is null")
				assert.Equal(t, "", result[0].test3, "result[0].test3 should be equal to \"\" as input is null")

				assert.Equal(t, 246, result[1].test, "v[1].test should be equal to 246")
				assert.Equal(t, -246, result[1].test2, "v[1].test2 should be equal to -247")
				assert.Equal(t, "string", result[1].test3, "v[1].test3 should be equal to \"string\"")

				assert.Equal(t, 247, result[2].test, "result[2].test should be equal to 246")
				assert.Equal(t, 248, result[2].test2, "result[2].test2 should be equal to 248")
				assert.Equal(t, "string", result[2].test3, "result[2].test3 should be equal to \"string\"")
			},
		},
		{
			name: "Stream test invalid JSON",
			streamReader: &StreamReader{
				readChan: make(chan string),
				done:     make(chan struct{}),
				data: `
					invalid json
					{"test":246,"test2":-246,"test3":"string"}
					{"test":247,"test2":248,"test3":"string"}
				`,
			},
			expectations: func(err error, result []*TestObj, t *testing.T) {
				assert.NotNil(t, err, "err is not nil as JSON is invalid")
				assert.IsType(t, InvalidJSONError(""), err, "err is of type InvalidJSONError")
				assert.Equal(t, "Invalid JSON", err.Error(), "err message is Invalid JSON")
			},
		},
	}
	for _, testCase := range tests {
		testCase := testCase
		t.Run(testCase.name, func(t *testing.T) {
			t.Parallel()
			runStreamTestCaseObjects(t, testCase)
		})
	}
}

func runStreamTestCaseObjects(t *testing.T, testCase StreamTestObject) {
	// create our channel which will receive our objects
	testChan := ChannelStreamObjects(make(chan *TestObj))
	dec := Stream.NewDecoder(testCase.streamReader)
	// start decoding (will block the goroutine until something is written to the ReadWriter)
	go dec.DecodeStream(&testChan)
	// start writing to the ReadWriter
	go testCase.streamReader.Write()
	// prepare our result
	result := []*TestObj{}
loop:
	for {
		select {
		case v := <-testChan:
			result = append(result, v)
		case <-dec.Done():
			break loop
		}
	}
	testCase.expectations(dec.Err(), result, t)
}

type ChannelStreamObjects chan *TestObj

func (c *ChannelStreamObjects) UnmarshalStream(dec *StreamDecoder) error {
	obj := &TestObj{}
	if err := dec.AddObject(obj); err != nil {
		return err
	}
	*c <- obj
	return nil
}

// Strings
type StreamTestString struct {
	name         string
	streamReader *StreamReader
	expectations func(error, []*string, *testing.T)
}

func TestStreamDecodingStringsParallel(t *testing.T) {
	var tests = []StreamTestString{
		{
			name: "Stream strings basic",
			streamReader: &StreamReader{
				readChan: make(chan string),
				done:     make(chan struct{}),
				data: `
					"hello"
					"world"
					"!"
				`,
			},
			expectations: func(err error, result []*string, t *testing.T) {
				assert.Nil(t, err, "err should be nil")

				assert.Equal(t, "hello", *result[0], "v[0] should be equal to 'hello'")
				assert.Equal(t, "world", *result[1], "v[1] should be equal to 'world'")
				assert.Equal(t, "!", *result[2], "v[2] should be equal to '!'")
			},
		},
		{
			name: "Stream strings with null",
			streamReader: &StreamReader{
				readChan: make(chan string),
				done:     make(chan struct{}),
				data: `
					"hello"
					null
					"!"
				`,
			},
			expectations: func(err error, result []*string, t *testing.T) {
				assert.Nil(t, err, "err should be nil")

				assert.Equal(t, "hello", *result[0], "v[0] should be equal to 'hello'")
				assert.Equal(t, "", *result[1], "v[1] should be equal to ''")
				assert.Equal(t, "!", *result[2], "v[2] should be equal to '!'")
			},
		},
		{
			name: "Stream strings invalid JSON",
			streamReader: &StreamReader{
				readChan: make(chan string),
				done:     make(chan struct{}),
				data: `
					"hello"
					world
					"!"
				`,
			},
			expectations: func(err error, result []*string, t *testing.T) {
				assert.NotNil(t, err, "err should not be nil")

				assert.IsType(t, InvalidJSONError(""), err, "err is of type InvalidJSONError")
				assert.Equal(t, "Invalid JSON", err.Error(), "err message is Invalid JSON")
			},
		},
	}
	for _, testCase := range tests {
		testCase := testCase
		t.Run(testCase.name, func(t *testing.T) {
			t.Parallel()
			runStreamTestCaseStrings(t, testCase)
		})
	}
}

func runStreamTestCaseStrings(t *testing.T, testCase StreamTestString) {
	// create our channel which will receive our objects
	testChan := ChannelStreamStrings(make(chan *string))
	dec := Stream.NewDecoder(testCase.streamReader)
	// start decoding (will block the goroutine until something is written to the ReadWriter)
	go dec.DecodeStream(&testChan)
	// start writing to the ReadWriter
	go testCase.streamReader.Write()
	// prepare our result
	result := []*string{}
loop:
	for {
		select {
		case v := <-testChan:
			result = append(result, v)
		case <-dec.Done():
			break loop
		}
	}
	testCase.expectations(dec.Err(), result, t)
}

type ChannelStreamStrings chan *string

func (c *ChannelStreamStrings) UnmarshalStream(dec *StreamDecoder) error {
	str := ""
	if err := dec.AddString(&str); err != nil {
		return err
	}
	*c <- &str
	return nil
}

// StreamReader mocks a stream reading chunks of data
type StreamReader struct {
	writeCounter int
	readChan     chan string
	done         chan struct{}
	data         string
}

func (r *StreamReader) Write() {
	l := len(r.data)
	t := 4
	chunkSize := l / t
	carry := 0
	lastWrite := 0
	for r.writeCounter < t {
		time.Sleep(time.Duration(r.writeCounter*100) * time.Millisecond)
		currentChunkStart := (chunkSize) * r.writeCounter
		lastWrite := currentChunkStart + chunkSize
		r.readChan <- r.data[currentChunkStart:lastWrite]
		carry = l - lastWrite
		r.writeCounter++
	}
	if carry > 0 {
		r.readChan <- r.data[lastWrite:]
	}
	r.done <- struct{}{}
}

func (r *StreamReader) Read(b []byte) (int, error) {
	select {
	case v := <-r.readChan:
		n := copy(b, v)
		return n, nil
	case <-r.done:
		return 0, nil
	}
}

// Deadline test
func TestStreamDecodingDeadline(t *testing.T) {
	dec := Stream.NewDecoder(&StreamReader{})
	now := time.Now()
	dec.SetDeadline(now)
	deadline, _ := dec.Deadline()
	assert.Equal(t, now.String(), deadline.String(), "dec.now and now should be equal")
	assert.Equal(t, now.String(), dec.deadline.String(), "dec.now and now should be equal")
}

func TestStreamDecodingErrNotSet(t *testing.T) {
	dec := Stream.NewDecoder(&StreamReader{})
	assert.Nil(t, dec.Err(), "dec.Err should be nim")
}