william/th7
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

timeout save() and cleanup

Browse Source
This commit is contained in:
William Clark 2023-11-24 10:46:49 +00:00
parent d7f138b7b9
commit 627834c072
2 changed files with 19 additions and 13 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
db/influxdb.go
View File

@ -30,9 +30,6 @@ type InfluxDBAdapter struct {
mu sync.Mutex
client *influxdb3.Client
cfg config.Config
token string
bucket string
host string
}
func NewInfluxDBAdapter(config config.Config) (*InfluxDBAdapter, error) {
@ -67,14 +64,11 @@ func NewInfluxDBAdapter(config config.Config) (*InfluxDBAdapter, error) {
}
adapter.cfg = config
adapter.token = config.DB["token"].(string)
adapter.bucket = config.DB["bucket"].(string)
adapter.host = config.DB["host"].(string)
adapter.client, err = influxdb3.New(influxdb3.ClientConfig{
Host: adapter.host,
Token: adapter.token,
Database: adapter.bucket,
Host: config.DB["host"].(string),
Token: config.DB["token"].(string),
Database: config.DB["bucket"].(string),
})
if err != nil {
@ -120,12 +114,12 @@ func (ad *InfluxDBAdapter) Save(channels []core.Channel) error {
id, thermo, gain, offset, value))
}
initContext, cancel := context.WithTimeout(context.Background(), INFLUXDB_MAX_SAVE_DUR)
saveContext, cancel := context.WithTimeout(context.Background(), INFLUXDB_MAX_SAVE_DUR)
defer cancel()
fmt.Printf("[InfluxDB] logged %d channels OK\n", len(channels))
err = ad.client.Write(initContext, []byte(sb.String()))
err = ad.client.Write(saveContext, []byte(sb.String()))
if err != nil {
return err
}

16
db/sqlite3.go
View File

@ -1,6 +1,7 @@
package db
import (
"context"
"database/sql"
"errors"
"fmt"
@ -20,6 +21,11 @@ type SQLite3Adapter struct {
cfg config.Config
}
const (
SQLITE3_MAX_INIT_DUR = 5000 * time.Millisecond
SQLITE3_MAX_SAVE_DUR = 2000 * time.Millisecond
)
func NewSQLite3Adapter(config config.Config) (*SQLite3Adapter, error) {
var adapter SQLite3Adapter
@ -39,6 +45,9 @@ func NewSQLite3Adapter(config config.Config) (*SQLite3Adapter, error) {
adapter.db = db
adapter.cfg = config
initContext, cancel := context.WithTimeout(context.Background(), SQLITE3_MAX_INIT_DUR)
defer cancel()
const create string = `
CREATE TABLE IF NOT EXISTS logs (
id INTEGER NOT NULL,
@ -49,7 +58,7 @@ func NewSQLite3Adapter(config config.Config) (*SQLite3Adapter, error) {
timestamp TEXT NOT NULL
);`
if _, err := adapter.db.Exec(create); err != nil {
if _, err := adapter.db.ExecContext(initContext, create); err != nil {
return &adapter, err
}
@ -72,6 +81,9 @@ func (ad *SQLite3Adapter) Save(channels []core.Channel) error {
timestamp := time.Now().Format(time.DateTime)
saveContext, cancel := context.WithTimeout(context.Background(), INFLUXDB_MAX_SAVE_DUR)
defer cancel()
for c := range channels {
id := channels[c].Id
@ -80,7 +92,7 @@ func (ad *SQLite3Adapter) Save(channels []core.Channel) error {
offset := ad.cfg.Channels[c].Offset
thermo := thermocouple.ThermoStringLookup[ad.cfg.Channels[c].Thermo]
_, err = statement.Exec(id, thermo, gain, offset, value, timestamp)
_, err = statement.ExecContext(saveContext, id, thermo, gain, offset, value, timestamp)
if err != nil {
return err
}