louis/oktopus
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(controller): implement transactions in database

Browse Source
This commit is contained in:
leandrofars 2024-02-18 12:08:26 -03:00
parent 8d10205878
commit 921a869ca1
1 changed files with 39 additions and 21 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

60
backend/services/controller/internal/db/device.go
View File

@ -45,33 +45,51 @@ func (d *Database) CreateDevice(device Device) error {
/* ------------------ Do not overwrite status of other mtp ------------------ */ /* ------------------ Do not overwrite status of other mtp ------------------ */
err := d.devices.FindOne(d.ctx, bson.D{{"sn", device.SN}}, nil).Decode(&deviceExistent) err := d.devices.FindOne(d.ctx, bson.D{{"sn", device.SN}}, nil).Decode(&deviceExistent)
if err != nil && err != mongo.ErrNoDocuments { if err == nil {
log.Println(err) if deviceExistent.Mqtt == Online {
return err device.Mqtt = Online
} }
if deviceExistent.Stomp == Online {
if deviceExistent.Mqtt == Online { device.Stomp = Online
device.Mqtt = Online }
} if deviceExistent.Websockets == Online {
if deviceExistent.Stomp == Online { device.Websockets = Online
device.Stomp = Online }
} } else {
if deviceExistent.Websockets == Online { if err != nil && err != mongo.ErrNoDocuments {
device.Websockets = Online log.Println(err)
return err
}
} }
/* -------------------------------------------------------------------------- */ /* -------------------------------------------------------------------------- */
opts := options.FindOneAndReplace().SetUpsert(true) callback := func(sessCtx mongo.SessionContext) (interface{}, error) {
// Important: You must pass sessCtx as the Context parameter to the operations for them to be executed in the
// transaction.
opts := options.FindOneAndReplace().SetUpsert(true)
err = d.devices.FindOneAndReplace(d.ctx, bson.D{{"sn", device.SN}}, device, opts).Decode(&result) err = d.devices.FindOneAndReplace(d.ctx, bson.D{{"sn", device.SN}}, device, opts).Decode(&result)
if err != nil { if err != nil {
if err == mongo.ErrNoDocuments { if err == mongo.ErrNoDocuments {
log.Printf("New device %s added to database", device.SN) log.Printf("New device %s added to database", device.SN)
return nil return nil, nil
}
return nil, err
} }
log.Fatal(err) log.Printf("Device %s already existed, and got replaced for new info", device.SN)
return nil, nil
}
session, err := d.client.StartSession()
if err != nil {
return err
}
defer session.EndSession(d.ctx)
_, err = session.WithTransaction(d.ctx, callback)
if err != nil {
return err
} }
log.Printf("Device %s already existed, and got replaced for new info", device.SN)
return err return err
} }
func (d *Database) RetrieveDevices(filter bson.A) ([]Device, error) { func (d *Database) RetrieveDevices(filter bson.A) ([]Device, error) {