cubefs

1
package sarama
2

3
import (
4
	"bytes"
5
	"compress/gzip"
6
	"fmt"
7
	"io"
8
	"sync"
9

10
	snappy "github.com/eapache/go-xerial-snappy"
11
	"github.com/pierrec/lz4"
12
)
13

14
var (
15
	lz4ReaderPool = sync.Pool{
16
		New: func() interface{} {
17
			return lz4.NewReader(nil)
18
		},
19
	}
20

21
	gzipReaderPool sync.Pool
22
)
23

24
func decompress(cc CompressionCodec, data []byte) ([]byte, error) {
25
	switch cc {
26
	case CompressionNone:
27
		return data, nil
28
	case CompressionGZIP:
29
		var err error
30
		reader, ok := gzipReaderPool.Get().(*gzip.Reader)
31
		if !ok {
32
			reader, err = gzip.NewReader(bytes.NewReader(data))
33
		} else {
34
			err = reader.Reset(bytes.NewReader(data))
35
		}
36

37
		if err != nil {
38
			return nil, err
39
		}
40

41
		defer gzipReaderPool.Put(reader)
42

43
		return io.ReadAll(reader)
44
	case CompressionSnappy:
45
		return snappy.Decode(data)
46
	case CompressionLZ4:
47
		reader, ok := lz4ReaderPool.Get().(*lz4.Reader)
48
		if !ok {
49
			reader = lz4.NewReader(bytes.NewReader(data))
50
		} else {
51
			reader.Reset(bytes.NewReader(data))
52
		}
53
		defer lz4ReaderPool.Put(reader)
54

55
		return io.ReadAll(reader)
56
	case CompressionZSTD:
57
		return zstdDecompress(ZstdDecoderParams{}, nil, data)
58
	default:
59
		return nil, PacketDecodingError{fmt.Sprintf("invalid compression specified (%d)", cc)}
60
	}
61
}
62