package transport import ( "bytes" "compress/gzip" "encoding/json" "fmt" "io" "sync" ) // SerializationFormat defines supported serialization formats type SerializationFormat string const ( SerializationJSON SerializationFormat = "json" SerializationMsgPack SerializationFormat = "msgpack" SerializationProtobuf SerializationFormat = "protobuf" SerializationAvro SerializationFormat = "avro" ) // CompressionType defines supported compression algorithms type CompressionType string const ( CompressionNone CompressionType = "none" CompressionGZip CompressionType = "gzip" CompressionLZ4 CompressionType = "lz4" CompressionSnappy CompressionType = "snappy" ) // SerializationConfig configures serialization behavior type SerializationConfig struct { Format SerializationFormat Compression CompressionType Encryption bool Validation bool } // SerializedMessage represents a serialized message with metadata type SerializedMessage struct { Format SerializationFormat `json:"format"` Compression CompressionType `json:"compression"` Encrypted bool `json:"encrypted"` Checksum string `json:"checksum"` Data []byte `json:"data"` Size int `json:"size"` Timestamp int64 `json:"timestamp"` } // Serializer interface defines serialization operations type Serializer interface { Serialize(msg *Message) (*SerializedMessage, error) Deserialize(serialized *SerializedMessage) (*Message, error) GetFormat() SerializationFormat GetConfig() SerializationConfig } // SerializationLayer manages multiple serializers and format selection type SerializationLayer struct { serializers map[SerializationFormat]Serializer defaultFormat SerializationFormat compressor Compressor encryptor Encryptor validator Validator mu sync.RWMutex } // Compressor interface for data compression type Compressor interface { Compress(data []byte, algorithm CompressionType) ([]byte, error) Decompress(data []byte, algorithm CompressionType) ([]byte, error) GetSupportedAlgorithms() []CompressionType } // Encryptor interface for data encryption type Encryptor interface { Encrypt(data []byte) ([]byte, error) Decrypt(data []byte) ([]byte, error) IsEnabled() bool } // Validator interface for data validation type Validator interface { Validate(msg *Message) error GenerateChecksum(data []byte) string VerifyChecksum(data []byte, checksum string) bool } // NewSerializationLayer creates a new serialization layer func NewSerializationLayer() *SerializationLayer { sl := &SerializationLayer{ serializers: make(map[SerializationFormat]Serializer), defaultFormat: SerializationJSON, compressor: NewDefaultCompressor(), validator: NewDefaultValidator(), } // Register default serializers sl.RegisterSerializer(NewJSONSerializer()) return sl } // RegisterSerializer registers a new serializer func (sl *SerializationLayer) RegisterSerializer(serializer Serializer) { sl.mu.Lock() defer sl.mu.Unlock() sl.serializers[serializer.GetFormat()] = serializer } // SetDefaultFormat sets the default serialization format func (sl *SerializationLayer) SetDefaultFormat(format SerializationFormat) { sl.mu.Lock() defer sl.mu.Unlock() sl.defaultFormat = format } // SetCompressor sets the compression handler func (sl *SerializationLayer) SetCompressor(compressor Compressor) { sl.mu.Lock() defer sl.mu.Unlock() sl.compressor = compressor } // SetEncryptor sets the encryption handler func (sl *SerializationLayer) SetEncryptor(encryptor Encryptor) { sl.mu.Lock() defer sl.mu.Unlock() sl.encryptor = encryptor } // SetValidator sets the validation handler func (sl *SerializationLayer) SetValidator(validator Validator) { sl.mu.Lock() defer sl.mu.Unlock() sl.validator = validator } // Serialize serializes a message using the specified or default format func (sl *SerializationLayer) Serialize(msg *Message, format ...SerializationFormat) (*SerializedMessage, error) { sl.mu.RLock() defer sl.mu.RUnlock() // Determine format to use selectedFormat := sl.defaultFormat if len(format) > 0 { selectedFormat = format[0] } // Get serializer serializer, exists := sl.serializers[selectedFormat] if !exists { return nil, fmt.Errorf("unsupported serialization format: %s", selectedFormat) } // Validate message if validator is configured if sl.validator != nil { if err := sl.validator.Validate(msg); err != nil { return nil, fmt.Errorf("message validation failed: %w", err) } } // Serialize message serialized, err := serializer.Serialize(msg) if err != nil { return nil, fmt.Errorf("serialization failed: %w", err) } // Apply compression if configured config := serializer.GetConfig() if config.Compression != CompressionNone && sl.compressor != nil { compressed, err := sl.compressor.Compress(serialized.Data, config.Compression) if err != nil { return nil, fmt.Errorf("compression failed: %w", err) } serialized.Data = compressed serialized.Compression = config.Compression } // Apply encryption if configured if config.Encryption && sl.encryptor != nil && sl.encryptor.IsEnabled() { encrypted, err := sl.encryptor.Encrypt(serialized.Data) if err != nil { return nil, fmt.Errorf("encryption failed: %w", err) } serialized.Data = encrypted serialized.Encrypted = true } // Generate checksum if sl.validator != nil { serialized.Checksum = sl.validator.GenerateChecksum(serialized.Data) } // Update metadata serialized.Size = len(serialized.Data) serialized.Timestamp = msg.Timestamp.UnixNano() return serialized, nil } // Deserialize deserializes a message func (sl *SerializationLayer) Deserialize(serialized *SerializedMessage) (*Message, error) { sl.mu.RLock() defer sl.mu.RUnlock() // Verify checksum if available if sl.validator != nil && serialized.Checksum != "" { if !sl.validator.VerifyChecksum(serialized.Data, serialized.Checksum) { return nil, fmt.Errorf("checksum verification failed") } } data := serialized.Data // Apply decryption if needed if serialized.Encrypted && sl.encryptor != nil && sl.encryptor.IsEnabled() { decrypted, err := sl.encryptor.Decrypt(data) if err != nil { return nil, fmt.Errorf("decryption failed: %w", err) } data = decrypted } // Apply decompression if needed if serialized.Compression != CompressionNone && sl.compressor != nil { decompressed, err := sl.compressor.Decompress(data, serialized.Compression) if err != nil { return nil, fmt.Errorf("decompression failed: %w", err) } data = decompressed } // Get serializer serializer, exists := sl.serializers[serialized.Format] if !exists { return nil, fmt.Errorf("unsupported serialization format: %s", serialized.Format) } // Create temporary serialized message for deserializer tempSerialized := &SerializedMessage{ Format: serialized.Format, Data: data, } // Deserialize message msg, err := serializer.Deserialize(tempSerialized) if err != nil { return nil, fmt.Errorf("deserialization failed: %w", err) } // Validate deserialized message if validator is configured if sl.validator != nil { if err := sl.validator.Validate(msg); err != nil { return nil, fmt.Errorf("deserialized message validation failed: %w", err) } } return msg, nil } // GetSupportedFormats returns all supported serialization formats func (sl *SerializationLayer) GetSupportedFormats() []SerializationFormat { sl.mu.RLock() defer sl.mu.RUnlock() formats := make([]SerializationFormat, 0, len(sl.serializers)) for format := range sl.serializers { formats = append(formats, format) } return formats } // JSONSerializer implements JSON serialization type JSONSerializer struct { config SerializationConfig } // NewJSONSerializer creates a new JSON serializer func NewJSONSerializer() *JSONSerializer { return &JSONSerializer{ config: SerializationConfig{ Format: SerializationJSON, Compression: CompressionNone, Encryption: false, Validation: true, }, } } // SetConfig updates the serializer configuration func (js *JSONSerializer) SetConfig(config SerializationConfig) { js.config = config js.config.Format = SerializationJSON // Ensure format is correct } // Serialize serializes a message to JSON func (js *JSONSerializer) Serialize(msg *Message) (*SerializedMessage, error) { data, err := json.Marshal(msg) if err != nil { return nil, fmt.Errorf("JSON marshal failed: %w", err) } return &SerializedMessage{ Format: SerializationJSON, Compression: CompressionNone, Encrypted: false, Data: data, Size: len(data), }, nil } // Deserialize deserializes a message from JSON func (js *JSONSerializer) Deserialize(serialized *SerializedMessage) (*Message, error) { var msg Message if err := json.Unmarshal(serialized.Data, &msg); err != nil { return nil, fmt.Errorf("JSON unmarshal failed: %w", err) } return &msg, nil } // GetFormat returns the serialization format func (js *JSONSerializer) GetFormat() SerializationFormat { return SerializationJSON } // GetConfig returns the serializer configuration func (js *JSONSerializer) GetConfig() SerializationConfig { return js.config } // DefaultCompressor implements basic compression operations type DefaultCompressor struct { supportedAlgorithms []CompressionType } // NewDefaultCompressor creates a new default compressor func NewDefaultCompressor() *DefaultCompressor { return &DefaultCompressor{ supportedAlgorithms: []CompressionType{ CompressionNone, CompressionGZip, }, } } // Compress compresses data using the specified algorithm func (dc *DefaultCompressor) Compress(data []byte, algorithm CompressionType) ([]byte, error) { switch algorithm { case CompressionNone: return data, nil case CompressionGZip: var buf bytes.Buffer writer := gzip.NewWriter(&buf) if _, err := writer.Write(data); err != nil { return nil, fmt.Errorf("gzip write failed: %w", err) } if err := writer.Close(); err != nil { return nil, fmt.Errorf("gzip close failed: %w", err) } return buf.Bytes(), nil default: return nil, fmt.Errorf("unsupported compression algorithm: %s", algorithm) } } // Decompress decompresses data using the specified algorithm func (dc *DefaultCompressor) Decompress(data []byte, algorithm CompressionType) ([]byte, error) { switch algorithm { case CompressionNone: return data, nil case CompressionGZip: reader, err := gzip.NewReader(bytes.NewReader(data)) if err != nil { return nil, fmt.Errorf("gzip reader creation failed: %w", err) } defer reader.Close() decompressed, err := io.ReadAll(reader) if err != nil { return nil, fmt.Errorf("gzip read failed: %w", err) } return decompressed, nil default: return nil, fmt.Errorf("unsupported compression algorithm: %s", algorithm) } } // GetSupportedAlgorithms returns supported compression algorithms func (dc *DefaultCompressor) GetSupportedAlgorithms() []CompressionType { return dc.supportedAlgorithms } // DefaultValidator implements basic message validation type DefaultValidator struct { strictMode bool } // NewDefaultValidator creates a new default validator func NewDefaultValidator() *DefaultValidator { return &DefaultValidator{ strictMode: false, } } // SetStrictMode enables/disables strict validation func (dv *DefaultValidator) SetStrictMode(enabled bool) { dv.strictMode = enabled } // Validate validates a message func (dv *DefaultValidator) Validate(msg *Message) error { if msg == nil { return fmt.Errorf("message is nil") } if msg.ID == "" { return fmt.Errorf("message ID is empty") } if msg.Topic == "" { return fmt.Errorf("message topic is empty") } if msg.Source == "" { return fmt.Errorf("message source is empty") } if msg.Type == "" { return fmt.Errorf("message type is empty") } if dv.strictMode { if msg.Data == nil { return fmt.Errorf("message data is nil") } if msg.Timestamp.IsZero() { return fmt.Errorf("message timestamp is zero") } } return nil } // GenerateChecksum generates a simple checksum for data func (dv *DefaultValidator) GenerateChecksum(data []byte) string { // Simple checksum implementation // In production, use a proper hash function like SHA-256 var sum uint32 for _, b := range data { sum += uint32(b) } return fmt.Sprintf("%08x", sum) } // VerifyChecksum verifies a checksum func (dv *DefaultValidator) VerifyChecksum(data []byte, checksum string) bool { return dv.GenerateChecksum(data) == checksum } // NoOpEncryptor implements a no-operation encryptor for testing type NoOpEncryptor struct { enabled bool } // NewNoOpEncryptor creates a new no-op encryptor func NewNoOpEncryptor() *NoOpEncryptor { return &NoOpEncryptor{enabled: false} } // SetEnabled enables/disables the encryptor func (noe *NoOpEncryptor) SetEnabled(enabled bool) { noe.enabled = enabled } // Encrypt returns data unchanged func (noe *NoOpEncryptor) Encrypt(data []byte) ([]byte, error) { return data, nil } // Decrypt returns data unchanged func (noe *NoOpEncryptor) Decrypt(data []byte) ([]byte, error) { return data, nil } // IsEnabled returns whether encryption is enabled func (noe *NoOpEncryptor) IsEnabled() bool { return noe.enabled } // SerializationMetrics tracks serialization performance type SerializationMetrics struct { SerializedMessages int64 DeserializedMessages int64 SerializationErrors int64 CompressionRatio float64 AverageMessageSize int64 TotalDataProcessed int64 } // MetricsCollector collects serialization metrics type MetricsCollector struct { metrics SerializationMetrics mu sync.RWMutex } // NewMetricsCollector creates a new metrics collector func NewMetricsCollector() *MetricsCollector { return &MetricsCollector{} } // RecordSerialization records a serialization operation func (mc *MetricsCollector) RecordSerialization(originalSize, serializedSize int) { mc.mu.Lock() defer mc.mu.Unlock() mc.metrics.SerializedMessages++ mc.metrics.TotalDataProcessed += int64(originalSize) // Update compression ratio if originalSize > 0 { ratio := float64(serializedSize) / float64(originalSize) mc.metrics.CompressionRatio = (mc.metrics.CompressionRatio + ratio) / 2 } // Update average message size mc.metrics.AverageMessageSize = mc.metrics.TotalDataProcessed / mc.metrics.SerializedMessages } // RecordDeserialization records a deserialization operation func (mc *MetricsCollector) RecordDeserialization() { mc.mu.Lock() defer mc.mu.Unlock() mc.metrics.DeserializedMessages++ } // RecordError records a serialization error func (mc *MetricsCollector) RecordError() { mc.mu.Lock() defer mc.mu.Unlock() mc.metrics.SerializationErrors++ } // GetMetrics returns current metrics func (mc *MetricsCollector) GetMetrics() SerializationMetrics { mc.mu.RLock() defer mc.mu.RUnlock() return mc.metrics } // Reset resets all metrics func (mc *MetricsCollector) Reset() { mc.mu.Lock() defer mc.mu.Unlock() mc.metrics = SerializationMetrics{} }