-
Notifications
You must be signed in to change notification settings - Fork 4
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
new db and repository method: ListenTable to listen for table's row c…
…hanges (INSERT, UPDATE, DELETE)
- Loading branch information
Showing
6 changed files
with
352 additions
and
14 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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,134 @@ | ||
| package pg | ||
|
|
||
| import ( | ||
| "context" | ||
| "encoding/json" | ||
| "fmt" | ||
| "sync/atomic" | ||
| ) | ||
|
|
||
| // TableChangeType is the type of the table change. | ||
| // Available values: INSERT, UPDATE, DELETE. | ||
| type TableChangeType string | ||
|
|
||
| const ( | ||
| // TableChangeTypeInsert is the INSERT table change type. | ||
| TableChangeTypeInsert TableChangeType = "INSERT" | ||
| // TableChangeTypeUpdate is the UPDATE table change type. | ||
| TableChangeTypeUpdate TableChangeType = "UPDATE" | ||
| // TableChangeTypeDelete is the DELETE table change type. | ||
| TableChangeTypeDelete TableChangeType = "DELETE" | ||
| ) | ||
|
|
||
| type ( | ||
| // TableNotification is the notification message sent by the postgresql server | ||
| // when a table change occurs. | ||
| // The subscribed postgres channel is named 'table_change_notifications'. | ||
| // The "old" and "new" fields are the old and new values of the row. | ||
| // The "old" field is only available for UPDATE and DELETE table change types. | ||
| // The "new" field is only available for INSERT and UPDATE table change types. | ||
| // The "old" and "new" fields are raw json values, use the "json.Unmarshal" to decode them. | ||
| // See "DB.ListenTable" method. | ||
| TableNotification[T any] struct { | ||
| Table string `json:"table"` | ||
| Change TableChangeType `json:"change"` // INSERT, UPDATE, DELETE. | ||
|
|
||
| New T `json:"new"` | ||
| Old T `json:"old"` | ||
| } | ||
|
|
||
| // TableNotificationJSON is the generic version of the TableNotification. | ||
| TableNotificationJSON = TableNotification[json.RawMessage] | ||
| ) | ||
|
|
||
| // ListenTable registers a function which notifies on the given "table" changes (INSERT, UPDATE, DELETE), | ||
| // the subscribed postgres channel is named 'table_change_notifications'. | ||
| // | ||
| // The callback function can return ErrStop to stop the listener without actual error. | ||
| // The callback function can return any other error to stop the listener and return the error. | ||
| // The callback function can return nil to continue listening. | ||
| // | ||
| // TableNotification's New and Old fields are raw json values, use the "json.Unmarshal" to decode them | ||
| // to the actual type. | ||
| func (db *DB) ListenTable(ctx context.Context, table string, callback func(TableNotificationJSON, error) error) (Closer, error) { | ||
| channelName := "table_change_notifications" | ||
|
|
||
| if atomic.LoadUint32(db.tableChangeNotifyFunctionOnce) == 0 { | ||
| // First, check and create the trigger for all tables. | ||
| query := fmt.Sprintf(` | ||
| CREATE OR REPLACE FUNCTION table_change_notify() RETURNS trigger AS $$ | ||
| DECLARE | ||
| payload text; | ||
| channel text := '%s'; | ||
| BEGIN | ||
| SELECT json_build_object('table', TG_TABLE_NAME, 'change', TG_OP, 'old', OLD, 'new', NEW)::text | ||
| INTO payload; | ||
| PERFORM pg_notify(channel, payload); | ||
| RETURN NEW; | ||
| END; | ||
| $$ | ||
| LANGUAGE plpgsql;`, channelName) | ||
|
|
||
| _, err := db.Exec(ctx, query) | ||
| if err != nil { | ||
| return nil, fmt.Errorf("create or replace function table_change_notify: %w", err) | ||
| } | ||
|
|
||
| atomic.StoreUint32(db.tableChangeNotifyFunctionOnce, 1) | ||
| } | ||
|
|
||
| db.tableChangeNotifyOnceMutex.RLock() | ||
| _, triggerCreated := db.tableChangeNotifyTriggerOnce[table] | ||
| db.tableChangeNotifyOnceMutex.RUnlock() | ||
| if !triggerCreated { | ||
| query := `CREATE TRIGGER ` + table + `_table_change_notify | ||
| BEFORE INSERT OR | ||
| UPDATE OR | ||
| DELETE | ||
| ON ` + table + ` | ||
| FOR EACH ROW | ||
| EXECUTE FUNCTION table_change_notify();` | ||
|
|
||
| _, err := db.Exec(ctx, query) | ||
| if err != nil { | ||
| return nil, fmt.Errorf("create trigger %s_table_change_notify: %w", table, err) | ||
| } | ||
|
|
||
| db.tableChangeNotifyOnceMutex.Lock() | ||
| db.tableChangeNotifyTriggerOnce[table] = struct{}{} | ||
| db.tableChangeNotifyOnceMutex.Unlock() | ||
| } | ||
|
|
||
| conn, err := db.Listen(ctx, channelName) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| go func() { | ||
| defer conn.Close(ctx) | ||
|
|
||
| for { | ||
| var evt TableNotificationJSON | ||
|
|
||
| notification, err := conn.Accept(context.Background()) | ||
| if err != nil { | ||
| if callback(evt, err) != nil { | ||
| return | ||
| } | ||
| } | ||
|
|
||
| if err = json.Unmarshal([]byte(notification.Payload), &evt); err != nil { | ||
| if callback(evt, err) != nil { | ||
| return | ||
| } | ||
| } | ||
|
|
||
| if err = callback(evt, nil); err != nil { | ||
| return | ||
| } | ||
| } | ||
| }() | ||
|
|
||
| return conn, nil | ||
| } |
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,58 @@ | ||
| package pg | ||
|
|
||
| import ( | ||
| "context" | ||
| "fmt" | ||
| "time" | ||
| ) | ||
|
|
||
| func ExampleDB_ListenTable() { | ||
| db, err := openTestConnection() | ||
| if err != nil { | ||
| handleExampleError(err) | ||
| return | ||
| } | ||
| defer db.Close() | ||
|
|
||
| closer, err := db.ListenTable(context.Background(), "customers", func(evt TableNotificationJSON, err error) error { | ||
| if err != nil { | ||
| fmt.Printf("received error: %v\n", err) | ||
| return err | ||
| } | ||
|
|
||
| if evt.Change == "INSERT" { | ||
| fmt.Printf("table: %s, event: %s, old: %s\n", evt.Table, evt.Change, string(evt.Old)) // new can't be predicated through its ID and timestamps. | ||
| } else { | ||
| fmt.Printf("table: %s, event: %s\n", evt.Table, evt.Change) | ||
| } | ||
|
|
||
| return nil | ||
| }) | ||
| if err != nil { | ||
| fmt.Println(err) | ||
| return | ||
| } | ||
| defer closer.Close(context.Background()) | ||
|
|
||
| newCustomer := Customer{ | ||
| CognitoUserID: "766064d4-a2a7-442d-aa75-33493bb4dbb9", | ||
| Email: "[email protected]", | ||
| Name: "Makis", | ||
| } | ||
| err = db.InsertSingle(context.Background(), newCustomer, &newCustomer.ID) | ||
| if err != nil { | ||
| fmt.Println(err) | ||
| return | ||
| } | ||
|
|
||
| newCustomer.Name = "Makis_UPDATED" | ||
| _, err = db.UpdateOnlyColumns(context.Background(), []string{"name"}, newCustomer) | ||
| if err != nil { | ||
| fmt.Println(err) | ||
| return | ||
| } | ||
| time.Sleep(5 * time.Second) // give it sometime to receive the notifications. | ||
| // Output: | ||
| // table: customers, event: INSERT, old: null | ||
| // table: customers, event: UPDATE | ||
| } |
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
Oops, something went wrong.