This repository has been archived by the owner on Feb 20, 2023. It is now read-only.
-
Notifications
You must be signed in to change notification settings - Fork 2
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #12 from cybertec-postgresql/11_new_record_state_e…
…xtraction Implement new record state extraction
- Loading branch information
Showing
12 changed files
with
419 additions
and
157 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,118 @@ | ||
| package kafka | ||
|
|
||
| import ( | ||
| "encoding/json" | ||
| "errors" | ||
| "strings" | ||
|
|
||
| kafka "github.com/segmentio/kafka-go" | ||
| ) | ||
|
|
||
| type cdcField struct { | ||
| Type string `json:"type"` | ||
| Optional bool `json:"optional"` | ||
| Field string `json:"field"` | ||
| } | ||
|
|
||
| type cdcFields struct { | ||
| Type string `json:"type"` | ||
| Fields []cdcField `json:"fields,omitempty"` | ||
| Optional bool `json:"optional"` | ||
| Name string `json:"name,omitempty"` | ||
| Field string `json:"field"` | ||
| } | ||
|
|
||
| type cdcSchema struct { | ||
| Type string `json:"type"` | ||
| Name string `json:"name"` | ||
| Fields []cdcFields `json:"fields"` | ||
| Optional bool `json:"optional"` | ||
| } | ||
|
|
||
| type cdcMessage struct { | ||
| Schema *cdcSchema `json:"schema"` | ||
| Payload *map[string]interface{} `json:"payload"` | ||
| } | ||
|
|
||
| type cdcKey struct { | ||
| Schema *cdcSchema `json:"schema"` | ||
| Payload *map[string]interface{} `json:"payload"` | ||
| } | ||
|
|
||
| // Message is a data structure representing kafka messages | ||
| type Message struct { | ||
| kafka.Message | ||
| Op string | ||
| TableName string | ||
| SchemaName string | ||
| Keys map[string]interface{} | ||
| Values map[string]interface{} | ||
| } | ||
|
|
||
| // NewMessage used to create and init a new message instance | ||
| func NewMessage(msg kafka.Message) (*Message, error) { | ||
| var err error | ||
| message := &Message{ | ||
| Message: msg, | ||
| Keys: make(map[string]interface{}), | ||
| Values: make(map[string]interface{}), | ||
| } | ||
| err = message.initKeys() | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
| err = message.initValues() | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
| return message, nil | ||
| } | ||
|
|
||
| // initKeys inits keys with the values to use in SQL DML statement | ||
| func (m *Message) initKeys() error { | ||
| var key cdcKey | ||
| if err := json.Unmarshal(m.Key, &key); err != nil { | ||
| return err | ||
| } | ||
| if key.Payload == nil { | ||
| return errors.New("Payload is nil") | ||
| } | ||
| m.Keys = *key.Payload | ||
| return nil | ||
| } | ||
|
|
||
| // initValues inits table name, operation and field names with the values to use in SQL DML statement | ||
| func (m *Message) initValues() error { | ||
| var msg cdcMessage | ||
| if err := json.Unmarshal(m.Value, &msg); err != nil { | ||
| return err | ||
| } | ||
| if msg.Payload == nil { | ||
| return errors.New("Payload is nil") | ||
| } | ||
| for k, v := range *msg.Payload { | ||
| if strings.HasPrefix(k, "__") { // system fields | ||
| switch k { | ||
| case "__schema": | ||
| m.SchemaName = v.(string) | ||
| case "__table": | ||
| m.TableName = v.(string) | ||
| case "__op": | ||
| m.Op = v.(string) | ||
| } | ||
| continue | ||
| } | ||
| m.Values[k] = v | ||
| } | ||
| return nil | ||
| } | ||
|
|
||
| func (m *Message) QualifiedTablename() string { | ||
| quoteIdent := func(s string) string { | ||
| return `"` + strings.Replace(s, `"`, `""`, -1) + `"` | ||
| } | ||
| if m.SchemaName > "" { | ||
| return quoteIdent(m.SchemaName) + "." + quoteIdent(m.TableName) | ||
| } | ||
| return quoteIdent(m.TableName) | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,46 @@ | ||
| package kafka | ||
|
|
||
| import ( | ||
| "testing" | ||
|
|
||
| kafka "github.com/segmentio/kafka-go" | ||
| "github.com/stretchr/testify/assert" | ||
| ) | ||
|
|
||
| func TestNewMessage(t *testing.T) { | ||
| m := kafka.Message{ | ||
| Value: []byte(`{"schema":{"type":"struct","fields":[{"type":"int32","optional":false,"field":"id"},{"type":"string","optional":false,"field":"first_name"},{"type":"string","optional":false,"field":"last_name"},{"type":"string","optional":false,"field":"email"},{"type":"int64","optional":true,"field":"__source_ts_ms"},{"type":"string","optional":true,"field":"__db"},{"type":"string","optional":true,"field":"__table"},{"type":"string","optional":true,"field":"__op"},{"type":"string","optional":true,"field":"__deleted"}],"optional":false,"name":"dbserver1.inventory.customers.Value"},"payload":{"id":1003,"first_name":"Edward","last_name":"Walker","email":"ed@walker.com","__source_ts_ms":0,"__db":"inventory","__schema":"public","__table":"customers","__op":"c","__deleted":"false"}}`), | ||
| Key: []byte(`{"schema":{"type":"struct","fields":[{"type":"int32","optional":false,"field":"id"}],"optional":false,"name":"dbserver1.inventory.customers.Key"},"payload":{"id":1003}}`), | ||
| } | ||
| msg, err := NewMessage(m) | ||
| assert.NoError(t, err) | ||
| assert.NotNil(t, msg) | ||
|
|
||
| m.Value = []byte(`{"schema":null, "payload":null}`) | ||
| msg, err = NewMessage(m) | ||
| assert.Error(t, err, "Corrupted value payload") | ||
| assert.Nil(t, msg) | ||
|
|
||
| m.Value = []byte{} | ||
| msg, err = NewMessage(m) | ||
| assert.Error(t, err, "Corrupted value") | ||
| assert.Nil(t, msg) | ||
|
|
||
| m.Key = []byte(`{"schema":null, "payload":null}`) | ||
| msg, err = NewMessage(m) | ||
| assert.Error(t, err, "Corrupted key payload") | ||
| assert.Nil(t, msg) | ||
|
|
||
| m.Key = []byte{} | ||
| msg, err = NewMessage(m) | ||
| assert.Error(t, err, "Corrupted key") | ||
| assert.Nil(t, msg) | ||
| } | ||
|
|
||
| func TestQualifiedTableName(t *testing.T) { | ||
| m := Message{} | ||
| m.TableName = "bar" | ||
| assert.Equal(t, m.QualifiedTablename(), `"bar"`, "No schema used") | ||
| m.SchemaName = "foo" | ||
| assert.Equal(t, m.QualifiedTablename(), `"foo"."bar"`, "Schema qualified") | ||
| } |
Oops, something went wrong.