refactors code

Author: raulb

pkg/processor/procbuiltin/unwrap.go

@@ -123,28 +123,12 @@ Example of an OpenCDC record:
 // openCDCUnwrapper unwraps an OpenCDC record from the payload, by unmarhsalling rec.Payload.After into type Record.
 type openCDCUnwrapper struct{}
 
-// Unwrap replaces the whole record.payload with record.payload.after.payload except position.
-func (o *openCDCUnwrapper) Unwrap(rec record.Record) (record.Record, error) {
-	var structData record.StructuredData
-	data := rec.Payload.After
-	switch d := data.(type) {
-	case record.RawData:
-		// unmarshal raw data to structured
-		err := json.Unmarshal(data.Bytes(), &structData)
-		if err != nil {
-			return record.Record{}, cerrors.Errorf("failed to unmarshal raw data as JSON: %w", unwrapProcType, err)
-		}
-	case record.StructuredData:
-		structData = d
-	default:
-		return record.Record{}, cerrors.Errorf("unexpected data type %T", unwrapProcType, data)
-	}
-
-	// get record.payload.after.operation
+// UnwrapOperation extracts operation from a structuredData record.
+func (o *openCDCUnwrapper) UnwrapOperation(structData record.StructuredData) (record.Operation, error) {
 	var operation record.Operation
 	op, ok := structData["operation"]
 	if !ok {
-		return record.Record{}, cerrors.Errorf("record payload after doesn't contain operation")
+		return operation, cerrors.Errorf("record payload after doesn't contain operation")
 	}
 
 	switch o := op.(type) {
@@ -153,15 +137,18 @@ func (o *openCDCUnwrapper) Unwrap(rec record.Record) (record.Record, error) {
 	case string:
 		err := operation.UnmarshalText([]byte(o))
 		if err != nil {
-			return record.Record{}, cerrors.Errorf("couldn't unmarshal record operation")
+			return operation, cerrors.Errorf("couldn't unmarshal record operation")
 		}
 	}
+	return operation, nil
+}
 
-	// get record.payload.after.metadata
+// UnwrapMetadata extracts metadata from a structuredData record.
+func (o *openCDCUnwrapper) UnwrapMetadata(structData record.StructuredData) (record.Metadata, error) {
 	var metadata record.Metadata
 	meta, ok := structData["metadata"]
 	if !ok {
-		return record.Record{}, cerrors.Errorf("record payload after doesn't contain metadata")
+		return metadata, cerrors.Errorf("record payload after doesn't contain metadata")
 	}
 
 	switch m := meta.(type) {
@@ -173,12 +160,15 @@ func (o *openCDCUnwrapper) Unwrap(rec record.Record) (record.Record, error) {
 			metadata[k] = fmt.Sprint(v)
 		}
 	}
+	return metadata, nil
+}
 
-	// get record.payload.after.key
+// UnwrapKey extracts key from a structuredData record.
+func (o *openCDCUnwrapper) UnwrapKey(structData record.StructuredData) (record.Data, error) {
 	var key record.Data
 	ky, ok := structData["key"]
 	if !ok {
-		return record.Record{}, cerrors.Errorf("record payload after doesn't contain key")
+		return key, cerrors.Errorf("record payload after doesn't contain key")
 	}
 
 	switch k := ky.(type) {
@@ -188,11 +178,15 @@ func (o *openCDCUnwrapper) Unwrap(rec record.Record) (record.Record, error) {
 		key = record.RawData{Raw: []byte(k)}
 	}
 
-	// get record.payload.after.payload.after
+	return key, nil
+}
+
+// UnwrapPayload extracts payload from a structuredData record.
+func (o *openCDCUnwrapper) UnwrapPayload(structData record.StructuredData) (record.Change, error) {
 	var payload record.Change
 	pl, ok := structData["payload"]
 	if !ok {
-		return record.Record{}, cerrors.Errorf("record payload doesn't contain payload")
+		return payload, cerrors.Errorf("record payload doesn't contain payload")
 	}
 
 	switch p := pl.(type) {
@@ -201,12 +195,12 @@ func (o *openCDCUnwrapper) Unwrap(rec record.Record) (record.Record, error) {
 	case map[string]interface{}:
 		afterData, ok := p["after"]
 		if !ok {
-			return record.Record{}, cerrors.Errorf("record payload after doesn't contain payload.after")
+			return payload, cerrors.Errorf("record payload after doesn't contain payload.after")
 		}
 
 		data, ok := afterData.(map[string]interface{})
 		if !ok {
-			return record.Record{}, cerrors.Errorf("record payload after payload.after is not a map")
+			return payload, cerrors.Errorf("record payload after payload.after is not a map")
 		}
 
 		convertedData := make(record.StructuredData, len(data))
@@ -219,6 +213,45 @@ func (o *openCDCUnwrapper) Unwrap(rec record.Record) (record.Record, error) {
 			After:  convertedData,
 		}
 	}
+	return payload, nil
+}
+
+// Unwrap replaces the whole record.payload with record.payload.after.payload except position.
+func (o *openCDCUnwrapper) Unwrap(rec record.Record) (record.Record, error) {
+	var structData record.StructuredData
+	data := rec.Payload.After
+	switch d := data.(type) {
+	case record.RawData:
+		// unmarshal raw data to structured
+		err := json.Unmarshal(data.Bytes(), &structData)
+		if err != nil {
+			return record.Record{}, cerrors.Errorf("failed to unmarshal raw data as JSON: %w", unwrapProcType, err)
+		}
+	case record.StructuredData:
+		structData = d
+	default:
+		return record.Record{}, cerrors.Errorf("unexpected data type %T", unwrapProcType, data)
+	}
+
+	operation, err := o.UnwrapOperation(structData)
+	if err != nil {
+		return record.Record{}, err
+	}
+
+	metadata, err := o.UnwrapMetadata(structData)
+	if err != nil {
+		return record.Record{}, err
+	}
+
+	key, err := o.UnwrapKey(structData)
+	if err != nil {
+		return record.Record{}, err
+	}
+
+	payload, err := o.UnwrapPayload(structData)
+	if err != nil {
+		return record.Record{}, err
+	}
 
 	return record.Record{
 		Key:       key,