mirror of
https://bitbucket.org/s_l_teichmann/mtsatellite
synced 2024-11-08 03:00:26 +01:00
Added experimental support for HKEYS to make minetestmapper happy.
This commit is contained in:
parent
720a6f497c
commit
d321d1888b
|
@ -8,6 +8,7 @@ type Session interface {
|
|||
Fetch(hash, key []byte) ([]byte, error)
|
||||
InTransaction() bool
|
||||
Store(hash, key, value []byte) (bool, error)
|
||||
AllKeys(hash []byte) (chan []byte, int, error)
|
||||
BeginTransaction() error
|
||||
CommitTransaction() error
|
||||
Close() error
|
||||
|
|
|
@ -99,6 +99,37 @@ func (c *Connection) Exec() bool {
|
|||
return c.writeBoolArray(arr)
|
||||
}
|
||||
|
||||
func (c *Connection) Hkeys(hash []byte) bool {
|
||||
var err error
|
||||
var n int
|
||||
var keys chan []byte
|
||||
if keys, n, err = c.session.AllKeys(hash); err != nil {
|
||||
return c.writeError(err)
|
||||
}
|
||||
|
||||
if err != nil {
|
||||
return c.writeError(err)
|
||||
}
|
||||
|
||||
if n == 0 {
|
||||
return c.writeEmptyArray()
|
||||
}
|
||||
|
||||
if _, err := c.conn.Write([]byte(fmt.Sprintf("*%d\r\n", n))); err != nil {
|
||||
logError(err)
|
||||
return false
|
||||
}
|
||||
|
||||
for key := range keys {
|
||||
if err := c.writeBulkString(key); err != nil {
|
||||
logError(err)
|
||||
close(keys)
|
||||
return false
|
||||
}
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
func (c *Connection) writeError(err error) bool {
|
||||
logError(err)
|
||||
if _, err = c.conn.Write(redisError); err != nil {
|
||||
|
|
38
leveldb.go
38
leveldb.go
|
@ -150,3 +150,41 @@ func (ldbs *LevelDBSession) CommitTransaction() (err error) {
|
|||
})
|
||||
return
|
||||
}
|
||||
|
||||
func (ldbs *LevelDBSession) AllKeys(hash []byte) (keys chan []byte, n int, err error) {
|
||||
ldbs.backend.mutex.RLock()
|
||||
|
||||
ro := leveldb.NewReadOptions()
|
||||
ro.SetFillCache(false)
|
||||
|
||||
it := ldbs.backend.db.NewIterator(ro)
|
||||
for ; it.Valid(); it.Next() {
|
||||
n++
|
||||
}
|
||||
|
||||
if err = it.GetError(); err != nil {
|
||||
ro.Close()
|
||||
it.Close()
|
||||
ldbs.backend.mutex.RUnlock()
|
||||
return
|
||||
}
|
||||
it.Close()
|
||||
|
||||
keys = make(chan []byte)
|
||||
|
||||
go func() {
|
||||
ldbs.backend.mutex.RUnlock()
|
||||
defer ro.Close()
|
||||
defer close(keys)
|
||||
it := ldbs.backend.db.NewIterator(ro)
|
||||
defer it.Close()
|
||||
for ; it.Valid(); it.Next() {
|
||||
keys <- it.Key()
|
||||
}
|
||||
if err := it.GetError(); err != nil {
|
||||
log.Printf("WARN: %s\n", err)
|
||||
}
|
||||
}()
|
||||
|
||||
return
|
||||
}
|
||||
|
|
13
parser.go
13
parser.go
|
@ -126,6 +126,7 @@ type RedisCommands interface {
|
|||
Hset(hash, key, block []byte) bool
|
||||
Multi() bool
|
||||
Exec() bool
|
||||
Hkeys(hash []byte) bool
|
||||
}
|
||||
|
||||
type RedisCommandExecutor struct {
|
||||
|
@ -204,6 +205,18 @@ func (rce *RedisCommandExecutor) execute() bool {
|
|||
|
||||
case "EXEC":
|
||||
return rce.commands.Exec()
|
||||
|
||||
case "HKEYS":
|
||||
if l < 2 {
|
||||
log.Println("WARN: Missing argments for HKEYS.")
|
||||
return false
|
||||
}
|
||||
hash, ok := rce.args[1].([]byte)
|
||||
if !ok {
|
||||
log.Println("WARN: HKEYS data are not byte slices.")
|
||||
return false
|
||||
}
|
||||
return rce.commands.Hkeys(hash)
|
||||
}
|
||||
log.Printf("WARN: unknown command: '%s'\n", cmd)
|
||||
return false
|
||||
|
|
50
sqlite.go
50
sqlite.go
|
@ -19,6 +19,8 @@ const (
|
|||
existsSql = "SELECT 1 FROM blocks WHERE pos = ?"
|
||||
updateSql = "UPDATE blocks SET data = ? WHERE pos = ?"
|
||||
insertSql = "INSERT INTO blocks (pos, data) VALUES (?, ?)"
|
||||
countSql = "SELECT count(*) FROM blocks"
|
||||
keysSql = "SELECT pos FROM blocks"
|
||||
)
|
||||
|
||||
type SqliteBackend struct {
|
||||
|
@ -27,6 +29,8 @@ type SqliteBackend struct {
|
|||
fetchStmt *sql.Stmt
|
||||
insertStmt *sql.Stmt
|
||||
updateStmt *sql.Stmt
|
||||
countStmt *sql.Stmt
|
||||
keysStmt *sql.Stmt
|
||||
}
|
||||
|
||||
type SqliteSession struct {
|
||||
|
@ -75,6 +79,16 @@ func NewSqliteBackend(path string) (sqlb *SqliteBackend, err error) {
|
|||
return
|
||||
}
|
||||
|
||||
if res.countStmt, err = res.db.Prepare(countSql); err != nil {
|
||||
res.closeAll()
|
||||
return
|
||||
}
|
||||
|
||||
if res.keysStmt, err = res.db.Prepare(keysSql); err != nil {
|
||||
res.closeAll()
|
||||
return
|
||||
}
|
||||
|
||||
sqlb = &res
|
||||
return
|
||||
}
|
||||
|
@ -102,6 +116,8 @@ func (sqlb *SqliteBackend) closeAll() error {
|
|||
closeStmt(&sqlb.insertStmt)
|
||||
closeStmt(&sqlb.updateStmt)
|
||||
closeStmt(&sqlb.existsStmt)
|
||||
closeStmt(&sqlb.countStmt)
|
||||
closeStmt(&sqlb.keysStmt)
|
||||
return closeDB(&sqlb.db)
|
||||
}
|
||||
|
||||
|
@ -198,3 +214,37 @@ func (ss *SqliteSession) CommitTransaction() error {
|
|||
ss.tx = nil
|
||||
return tx.Commit()
|
||||
}
|
||||
|
||||
func (ss *SqliteSession) AllKeys(hash []byte) (keys chan []byte, n int, err error) {
|
||||
globalLock.Lock()
|
||||
defer globalLock.Unlock()
|
||||
|
||||
countStmt := ss.txStmt(ss.backend.countStmt)
|
||||
if err = countStmt.QueryRow().Scan(&n); err != nil {
|
||||
if err == sql.ErrNoRows {
|
||||
err = nil
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
keysStmt := ss.txStmt(ss.backend.keysStmt)
|
||||
var rows *sql.Rows
|
||||
if rows, err = keysStmt.Query(); err != nil {
|
||||
return
|
||||
}
|
||||
keys = make(chan []byte)
|
||||
go func() {
|
||||
defer rows.Close()
|
||||
defer close(keys)
|
||||
for rows.Next() {
|
||||
var key []byte
|
||||
if err := rows.Scan(&key); err != nil {
|
||||
log.Printf("WARN: %s", err)
|
||||
break
|
||||
}
|
||||
keys <- key
|
||||
}
|
||||
}()
|
||||
|
||||
return
|
||||
}
|
||||
|
|
Loading…
Reference in New Issue
Block a user