// Copyright 2021 EMQ Technologies Co., Ltd. // // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. // You may obtain a copy of the License at // // http://www.apache.org/licenses/LICENSE-2.0 // // Unless required by applicable law or agreed to in writing, software // distributed under the License is distributed on an "AS IS" BASIS, // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. // See the License for the specific language governing permissions and // limitations under the License. package tskv import ( "bytes" "database/sql" "encoding/gob" "fmt" "github.com/lf-edge/ekuiper/internal/conf" _ "github.com/mattn/go-sqlite3" "path" "sync" ) // All TSKV instances share ONE database with different tables var ( db *sql.DB once sync.Once ) // SqliteTskv All TSKV instances share the same database but with different tables // Each table must have ONLY ONE instance type SqliteTskv struct { table string // only append key bigger than the latest key inside; ONLY check in the instance itself last int64 } func init() { gob.Register(make(map[string]interface{})) } func NewSqlite(table string) (*SqliteTskv, error) { var outerError error once.Do(func() { d, err := conf.GetDataLoc() if err != nil { outerError = err return } db, outerError = sql.Open("sqlite3", path.Join(d, "tskv.db")) }) if outerError != nil { return nil, outerError } if db == nil { return nil, fmt.Errorf("cannot initiate sqlite db, please restart") } sqlStr := fmt.Sprintf("CREATE TABLE IF NOT EXISTS '%s'('key' INTEGER PRIMARY KEY, 'val' BLOB);", table) _, outerError = db.Exec(sqlStr) if outerError != nil { return nil, fmt.Errorf("cannot create table: %v", outerError) } return &SqliteTskv{ table: table, last: last(table), }, nil } func (m *SqliteTskv) Set(key int64, value interface{}) (bool, error) { if key > m.last { b, err := m.encode(value) if err != nil { return false, err } sqlStr := fmt.Sprintf("INSERT INTO %s(key,val) values(?,?);", m.table) stmt, err := db.Prepare(sqlStr) if err != nil { return false, err } defer stmt.Close() _, err = stmt.Exec(key, b) if err != nil { return false, err } else { m.last = key return true, nil } } else { return false, nil } } func (m *SqliteTskv) Get(key int64, value interface{}) (bool, error) { sqlStr := fmt.Sprintf("SELECT val FROM %s WHERE key=%d;", m.table, key) row := db.QueryRow(sqlStr) var tmp []byte switch err := row.Scan(&tmp); err { case sql.ErrNoRows: return false, nil case nil: // do nothing, continue processing default: return false, err } dec := gob.NewDecoder(bytes.NewBuffer(tmp)) if err := dec.Decode(value); err != nil { return false, err } return true, nil } func (m *SqliteTskv) Last(value interface{}) (int64, error) { _, err := m.Get(m.last, value) if err != nil { return 0, err } return m.last, nil } func (m *SqliteTskv) Delete(k int64) error { sqlStr := fmt.Sprintf("DELETE FROM %s WHERE key=%d;", m.table, k) _, err := db.Exec(sqlStr) return err } func (m *SqliteTskv) DeleteBefore(k int64) error { sqlStr := fmt.Sprintf("DELETE FROM %s WHERE key<%d;", m.table, k) _, err := db.Exec(sqlStr) return err } func (m *SqliteTskv) Close() error { return nil } func (m *SqliteTskv) Drop() error { sqlStr := fmt.Sprintf("Drop table %s;", m.table) _, err := db.Exec(sqlStr) return err } func (m *SqliteTskv) encode(value interface{}) ([]byte, error) { var buf bytes.Buffer gob.Register(value) enc := gob.NewEncoder(&buf) if err := enc.Encode(value); err != nil { return nil, err } return buf.Bytes(), nil } func last(table string) int64 { sqlStr := fmt.Sprintf("SELECT key FROM %s Order by key DESC Limit 1;", table) row := db.QueryRow(sqlStr) var tmp int64 switch err := row.Scan(&tmp); err { case sql.ErrNoRows: return 0 case nil: return tmp default: return 0 } }