wal-g

1
package internal
2

3
import (
4
	"bytes"
5
	"encoding/json"
6
	"fmt"
7
	"io"
8

9
	"github.com/spf13/viper"
10
	streamJSON "github.com/wal-g/json"
11
	"github.com/wal-g/tracelog"
12
	conf "github.com/wal-g/wal-g/internal/config"
13
)
14

15
type UnknownSerializerTypeError struct {
16
	error
17
}
18

19
func NewUnknownSerializerTypeError(serializerType DtoSerializerType) UnknownSerializerTypeError {
20
	return UnknownSerializerTypeError{fmt.Errorf("undefined dto serializer type: %s", serializerType)}
21
}
22

23
func (err UnknownSerializerTypeError) Error() string {
24
	return fmt.Sprintf(tracelog.GetErrorFormatter(), err.error)
25
}
26

27
type DtoSerializerType string
28

29
const (
30
	RegularJSONSerializer  DtoSerializerType = "json_default"
31
	StreamedJSONSerializer DtoSerializerType = "json_streamed"
32
)
33

34
type DtoSerializer interface {
35
	Marshal(dto interface{}) (io.Reader, error)
36
	Unmarshal(reader io.Reader, dto interface{}) error
37
}
38

39
func NewDtoSerializer() (DtoSerializer, error) {
40
	switch t := DtoSerializerType(viper.GetString(conf.SerializerTypeSetting)); t {
41
	case RegularJSONSerializer:
42
		return RegularJSON{}, nil
43
	case StreamedJSONSerializer:
44
		return StreamedJSON{}, nil
45
	default:
46
		return nil, NewUnknownSerializerTypeError(t)
47
	}
48
}
49

50
var _ DtoSerializer = RegularJSON{}
51

52
type RegularJSON struct{}
53

54
func (r RegularJSON) Marshal(dto interface{}) (io.Reader, error) {
55
	data, err := json.Marshal(dto)
56
	if err != nil {
57
		return nil, err
58
	}
59
	return bytes.NewReader(data), nil
60
}
61

62
func (r RegularJSON) Unmarshal(reader io.Reader, dto interface{}) error {
63
	data, err := io.ReadAll(reader)
64
	if err != nil {
65
		return err
66
	}
67
	return json.Unmarshal(data, dto)
68
}
69

70
var _ DtoSerializer = StreamedJSON{}
71

72
type StreamedJSON struct{}
73

74
func (s StreamedJSON) Marshal(dto interface{}) (io.Reader, error) {
75
	r, w := io.Pipe()
76
	go func() {
77
		if err := streamJSON.Marshal(dto, w); err != nil {
78
			_ = w.CloseWithError(err)
79
		}
80
	}()
81
	return r, nil
82
}
83

84
func (s StreamedJSON) Unmarshal(reader io.Reader, dto interface{}) error {
85
	return streamJSON.Unmarshal(reader, dto)
86
}
87