add undo compressor log support (#659)

Author: smiletrl

pkg/compressor/compressor_factory.go

@@ -17,61 +17,37 @@
 
 package compressor
 
-type CompressorType int8
+type CompressorType string
 
 const (
-	CompressorNone CompressorType = iota
-	CompressorGzip
-	CompressorZip
-	CompressorSevenz
-	CompressorBzip2
-	CompressorLz4
-	CompressorDeflate
-	CompressorZstd
+	// "None" means no compressor is used
+	CompressorNone    CompressorType = "None"
+	CompressorGzip    CompressorType = "Gzip"
+	CompressorZip     CompressorType = "Zip"
+	CompressorSevenz  CompressorType = "Sevenz"
+	CompressorBzip2   CompressorType = "Bzip2"
+	CompressorLz4     CompressorType = "Lz4"
+	CompressorDeflate CompressorType = "Deflate"
+	CompressorZstd    CompressorType = "Zstd"
 )
 
-func (c CompressorType) String() string {
+func (c CompressorType) GetCompressor() Compressor {
 	switch c {
 	case CompressorNone:
-		return "CompressorNone"
+		return &NoneCompressor{}
 	case CompressorGzip:
-		return "CompressorGzip"
+		return &Gzip{}
 	case CompressorZip:
-		return "CompressorZip"
-	case CompressorSevenz:
-		return "CompressorSevenz"
+		return &Zip{}
 	case CompressorBzip2:
-		return "CompressorBzip2"
+		return &Bzip2{}
 	case CompressorLz4:
-		return "CompressorLz4"
+		return &Lz4{}
 	case CompressorZstd:
-		return "CompressorZstd"
+		return &Zstd{}
 	case CompressorDeflate:
-		return "CompressorDeflate"
+		return &DeflateCompress{}
 	default:
-		return ""
-	}
-}
-
-var compressor map[string]CompressorType
-
-func GetByName(name string) CompressorType {
-	if compressor == nil {
-		compressor = map[string]CompressorType{
-			CompressorNone.String():    CompressorNone,
-			CompressorGzip.String():    CompressorGzip,
-			CompressorZip.String():     CompressorZip,
-			CompressorSevenz.String():  CompressorSevenz,
-			CompressorBzip2.String():   CompressorBzip2,
-			CompressorLz4.String():     CompressorLz4,
-			CompressorZstd.String():    CompressorZstd,
-			CompressorDeflate.String(): CompressorDeflate,
-		}
-	}
-
-	if v, ok := compressor[name]; ok {
-		return v
-	} else {
-		return CompressorNone
+		return &NoneCompressor{}
 	}
 }

pkg/compressor/compressor_type.go

@@ -223,8 +223,7 @@ func (m *BaseUndoLogManager) FlushUndoLog(tranCtx *types.TransactionContext, con
 
 	parseContext := make(map[string]string, 0)
 	parseContext[serializerKey] = "json"
-	// Todo use config
-	parseContext[compressorTypeKey] = compressor.CompressorNone.String()
+	parseContext[compressorTypeKey] = undo.UndoConfig.CompressConfig.Type
 	undoLogContent := m.encodeUndoLogCtx(parseContext)
 	rollbackInfo, err := m.serializeBranchUndoLog(&branchUndoLog, parseContext[serializerKey])
 	if err != nil {
@@ -379,7 +378,7 @@ func (m *BaseUndoLogManager) insertUndoLogWithGlobalFinished(ctx context.Context
 	// todo use config to replace
 	parseContext := make(map[string]string, 0)
 	parseContext[serializerKey] = "json"
-	parseContext[compressorTypeKey] = compressor.CompressorNone.String()
+	parseContext[compressorTypeKey] = undo.UndoConfig.CompressConfig.Type
 	undoLogContent := m.encodeUndoLogCtx(parseContext)
 
 	logParse, err := parser.GetCache().Load(parseContext[serializerKey])
@@ -495,7 +494,7 @@ func (m *BaseUndoLogManager) getRollbackInfo(rollbackInfo []byte, undoContext ma
 	res := rollbackInfo
 	// get compress type
 	if v, ok := undoContext[compressorTypeKey]; ok {
-		res, err = compressor.GetByName(v).GetCompressor().Decompress(rollbackInfo)
+		res, err = compressor.CompressorType(v).GetCompressor().Decompress(rollbackInfo)
 		if err != nil {
 			log.Errorf("[getRollbackInfo] decompress fail, err: %+v", err)
 			return nil, err

pkg/compressor/compressor_type_test.go

@@ -19,6 +19,8 @@ package undo
 
 import (
 	"flag"
+
+	"github.com/seata/seata-go/pkg/compressor"
 )
 
 var (
@@ -52,7 +54,6 @@ func (u *Config) RegisterFlagsWithPrefix(prefix string, f *flag.FlagSet) {
 }
 
 func (c *CompressConfig) RegisterFlagsWithPrefix(prefix string, f *flag.FlagSet) {
-	f.BoolVar(&c.Enable, prefix+".enable", true, "Whether compression is required.")
-	f.StringVar(&c.Type, prefix+".type", "zip", "Compression type")
+	f.StringVar(&c.Type, prefix+".type", string(compressor.CompressorNone), "Compression type")
 	f.StringVar(&c.Threshold, prefix+".threshold", "64k", "Compression threshold")
 }