postgres-switch #113
@ -8,27 +8,18 @@ import (
|
|||||||
"git.defalsify.org/vise.git/lang"
|
"git.defalsify.org/vise.git/lang"
|
||||||
)
|
)
|
||||||
|
|
||||||
var (
|
type PostgresDb struct {
|
||||||
pdbC map[string]chan db.Db
|
|
||||||
)
|
|
||||||
|
|
||||||
type ThreadPostgresDb struct {
|
|
||||||
db db.Db
|
db db.Db
|
||||||
connStr string
|
connStr string
|
||||||
}
|
}
|
||||||
|
|
||||||
Alfred-mk marked this conversation as resolved
Outdated
|
|||||||
func NewThreadPostgresDb() *ThreadPostgresDb {
|
func NewPostgresDb() *PostgresDb {
|
||||||
if pdbC == nil {
|
return &PostgresDb{}
|
||||||
pdbC = make(map[string]chan db.Db)
|
|
||||||
}
|
|
||||||
return &ThreadPostgresDb{}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
func (tpdb *ThreadPostgresDb) Connect(ctx context.Context, connStr string) error {
|
func (pdb *PostgresDb) Connect(ctx context.Context, connStr string) error {
|
||||||
var ok bool
|
if pdb.db != nil {
|
||||||
_, ok = pdbC[connStr]
|
logg.WarnCtxf(ctx, "already connected, skipping", "connStr", connStr)
|
||||||
if ok {
|
|
||||||
logg.WarnCtxf(ctx, "already registered thread postgres, skipping", "connStr", connStr)
|
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
postgresdb := postgres.NewPgDb().WithSchema("public")
|
postgresdb := postgres.NewPgDb().WithSchema("public")
|
||||||
@ -36,80 +27,48 @@ func (tpdb *ThreadPostgresDb) Connect(ctx context.Context, connStr string) error
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
pdbC[connStr] = make(chan db.Db, 1)
|
pdb.db = postgresdb
|
||||||
pdbC[connStr] <- postgresdb
|
pdb.connStr = connStr
|
||||||
tpdb.connStr = connStr
|
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func (tpdb *ThreadPostgresDb) reserve() {
|
func (pdb *PostgresDb) SetPrefix(pfx uint8) {
|
||||||
if tpdb.db == nil {
|
pdb.db.SetPrefix(pfx)
|
||||||
tpdb.db = <-pdbC[tpdb.connStr]
|
}
|
||||||
|
|
||||||
|
func (pdb *PostgresDb) SetSession(sessionId string) {
|
||||||
|
pdb.db.SetSession(sessionId)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (pdb *PostgresDb) SetLanguage(lng *lang.Language) {
|
||||||
|
pdb.db.SetLanguage(lng)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (pdb *PostgresDb) Safe() bool {
|
||||||
|
return pdb.db.Safe()
|
||||||
|
}
|
||||||
|
|
||||||
|
func (pdb *PostgresDb) Prefix() uint8 {
|
||||||
|
return pdb.db.Prefix()
|
||||||
|
}
|
||||||
|
|
||||||
|
func (pdb *PostgresDb) SetLock(typ uint8, locked bool) error {
|
||||||
|
return pdb.db.SetLock(typ, locked)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (pdb *PostgresDb) Put(ctx context.Context, key []byte, val []byte) error {
|
||||||
|
return pdb.db.Put(ctx, key, val)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (pdb *PostgresDb) Get(ctx context.Context, key []byte) ([]byte, error) {
|
||||||
|
return pdb.db.Get(ctx, key)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (pdb *PostgresDb) Close() error {
|
||||||
|
if pdb.db == nil {
|
||||||
|
return nil
|
||||||
}
|
}
|
||||||
}
|
err := pdb.db.Close()
|
||||||
|
pdb.db = nil
|
||||||
func (tpdb *ThreadPostgresDb) release() {
|
|
||||||
if tpdb.db == nil {
|
|
||||||
return
|
|
||||||
}
|
|
||||||
pdbC[tpdb.connStr] <- tpdb.db
|
|
||||||
tpdb.db = nil
|
|
||||||
}
|
|
||||||
|
|
||||||
func (tpdb *ThreadPostgresDb) SetPrefix(pfx uint8) {
|
|
||||||
tpdb.reserve()
|
|
||||||
tpdb.db.SetPrefix(pfx)
|
|
||||||
}
|
|
||||||
|
|
||||||
func (tpdb *ThreadPostgresDb) SetSession(sessionId string) {
|
|
||||||
tpdb.reserve()
|
|
||||||
tpdb.db.SetSession(sessionId)
|
|
||||||
}
|
|
||||||
|
|
||||||
func (tpdb *ThreadPostgresDb) SetLanguage(lng *lang.Language) {
|
|
||||||
tpdb.reserve()
|
|
||||||
tpdb.db.SetLanguage(lng)
|
|
||||||
}
|
|
||||||
|
|
||||||
func (tpdb *ThreadPostgresDb) Safe() bool {
|
|
||||||
tpdb.reserve()
|
|
||||||
v := tpdb.db.Safe()
|
|
||||||
tpdb.release()
|
|
||||||
return v
|
|
||||||
}
|
|
||||||
|
|
||||||
func (tpdb *ThreadPostgresDb) Prefix() uint8 {
|
|
||||||
tpdb.reserve()
|
|
||||||
v := tpdb.db.Prefix()
|
|
||||||
tpdb.release()
|
|
||||||
return v
|
|
||||||
}
|
|
||||||
|
|
||||||
func (tpdb *ThreadPostgresDb) SetLock(typ uint8, locked bool) error {
|
|
||||||
tpdb.reserve()
|
|
||||||
err := tpdb.db.SetLock(typ, locked)
|
|
||||||
tpdb.release()
|
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
func (tpdb *ThreadPostgresDb) Put(ctx context.Context, key []byte, val []byte) error {
|
|
||||||
tpdb.reserve()
|
|
||||||
err := tpdb.db.Put(ctx, key, val)
|
|
||||||
tpdb.release()
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
|
|
||||||
func (tpdb *ThreadPostgresDb) Get(ctx context.Context, key []byte) ([]byte, error) {
|
|
||||||
tpdb.reserve()
|
|
||||||
v, err := tpdb.db.Get(ctx, key)
|
|
||||||
tpdb.release()
|
|
||||||
return v, err
|
|
||||||
}
|
|
||||||
|
|
||||||
func (tpdb *ThreadPostgresDb) Close() error {
|
|
||||||
tpdb.reserve()
|
|
||||||
close(pdbC[tpdb.connStr])
|
|
||||||
err := tpdb.db.Close()
|
|
||||||
tpdb.db = nil
|
|
||||||
return err
|
|
||||||
}
|
|
@ -72,7 +72,7 @@ func (ms *MenuStorageService) GetUserdataDb(ctx context.Context) (db.Db, error)
|
|||||||
}
|
}
|
||||||
|
|
||||||
if database == "postgres" {
|
if database == "postgres" {
|
||||||
Alfred-mk marked this conversation as resolved
Outdated
lash
commented
postgres shouold be implemented for both userdatatstore and state store postgres shouold be implemented for both userdatatstore and state store
|
|||||||
ms.userDataStore = NewThreadPostgresDb()
|
ms.userDataStore = NewPostgresDb()
|
||||||
connStr := buildConnStr()
|
connStr := buildConnStr()
|
||||||
err := ms.userDataStore.Connect(ctx, connStr)
|
err := ms.userDataStore.Connect(ctx, connStr)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
Loading…
Reference in New Issue
Block a user
i dont think you need the thread abstraction here, because the pg implementation used is already a managed connection pool. did you try it without?