1263 lines
34 KiB
V
1263 lines
34 KiB
V
// https://redis.io/docs/latest/develop/reference/protocol-spec/
|
|
|
|
module redis
|
|
|
|
import math.big
|
|
import net
|
|
import net.ssl
|
|
import strings
|
|
|
|
// RESP3 wrapper types
|
|
pub struct RedisBlobError {
|
|
pub:
|
|
data []u8
|
|
}
|
|
|
|
pub struct RedisVerbatim {
|
|
pub:
|
|
format string
|
|
data []u8
|
|
}
|
|
|
|
pub struct RedisMap {
|
|
pub:
|
|
// interleaved key/value pairs: [k1, v1, k2, v2, ...]
|
|
pairs []RedisValue
|
|
}
|
|
|
|
pub struct RedisSet {
|
|
pub:
|
|
elements []RedisValue
|
|
}
|
|
|
|
pub struct RedisPush {
|
|
pub:
|
|
elements []RedisValue
|
|
}
|
|
|
|
// RedisValue represents all possible RESP (Redis Serialization Protocol) data types
|
|
pub type RedisValue = bool
|
|
| big.Integer
|
|
| f32
|
|
| f64
|
|
| i64
|
|
| []u8
|
|
| RedisBlobError
|
|
| RedisMap
|
|
| RedisNull
|
|
| RedisPush
|
|
| RedisSet
|
|
| map[string]RedisValue
|
|
| []RedisValue
|
|
| RedisVerbatim
|
|
| string
|
|
|
|
// RedisNull represents the Redis NULL type
|
|
pub struct RedisNull {}
|
|
|
|
const cmd_buf_pre_allocate_len = 4096 // Initial buffer size for command building
|
|
const resp_buf_pre_allocate_len = 8192 // Initial buffer size for response reading
|
|
const max_skip = 64 // Max non-prefix bytes to skip when resynchronizing
|
|
|
|
// DB represents a Redis database connection
|
|
pub struct DB {
|
|
pub mut:
|
|
version int // RESP protocol version
|
|
conn &net.TcpConn = unsafe { nil } // TCP connection to Redis
|
|
ssl_conn &ssl.SSLConn = unsafe { nil } // SSL connection to Redis
|
|
tls bool
|
|
|
|
// Pre-allocated buffers to reduce memory allocations
|
|
cmd_buf []u8 // Buffer for building commands
|
|
resp_buf []u8 // Buffer for reading responses
|
|
pipeline_mode bool
|
|
pipeline_buffer []u8
|
|
pipeline_cmd_count int
|
|
}
|
|
|
|
// Configuration options for Redis connection
|
|
@[params]
|
|
pub struct Config {
|
|
pub mut:
|
|
host string = '127.0.0.1' // Redis server host
|
|
port u16 = 6379 // Redis server port
|
|
password string // Redis server password (optional)
|
|
tls bool // Enable TLS/SSL connection
|
|
version int @[deprecated] // ignored - RESP version auto handled in connect
|
|
}
|
|
|
|
fn resp2_auth(password string, mut db DB) ! {
|
|
db.version = 2
|
|
if password.len > 0 {
|
|
db.auth(password)!
|
|
}
|
|
}
|
|
|
|
// connect establishes a connection to a Redis server
|
|
pub fn connect(config Config) !DB {
|
|
mut db := DB{
|
|
tls: config.tls
|
|
cmd_buf: []u8{cap: cmd_buf_pre_allocate_len}
|
|
resp_buf: []u8{cap: resp_buf_pre_allocate_len}
|
|
}
|
|
|
|
if config.tls {
|
|
mut ssl_conn := ssl.new_ssl_conn(ssl.SSLConnectConfig{ validate: false })!
|
|
ssl_conn.dial(config.host, int(config.port))!
|
|
db.ssl_conn = ssl_conn
|
|
} else {
|
|
db.conn = net.dial_tcp('${config.host}:${config.port}')!
|
|
}
|
|
|
|
// Always attempt HELLO during connect() to negotiate RESP3 (and include AUTH
|
|
// subcommand when a password is provided). If HELLO fails or the server
|
|
// doesn't support it, fall back to RESP2 and perform AUTH via the AUTH
|
|
// command (if a password was given).
|
|
|
|
// default version 3
|
|
db.version = 3
|
|
|
|
// build HELLO 3 command; include AUTH subcommand when password present
|
|
db.cmd_buf.clear()
|
|
if config.password.len > 0 {
|
|
// *4\r\n$5\r\nHELLO\r\n$1\r\n3\r\n$4\r\nAUTH\r\n$<pwlen>\r\n<pw>\r\n
|
|
db.cmd_buf << '*4\r\n$5\r\nHELLO\r\n$1\r\n3\r\n$4\r\nAUTH\r\n$${config.password.len}\r\n${config.password}\r\n'.bytes()
|
|
} else {
|
|
// *2\r\n$5\r\nHELLO\r\n$1\r\n3\r\n
|
|
db.cmd_buf << '*2\r\n$5\r\nHELLO\r\n$1\r\n3\r\n'.bytes()
|
|
}
|
|
|
|
// send HELLO and attempt to read response. If any step fails, fallback to RESP2 + AUTH.
|
|
db.write_data(db.cmd_buf) or {
|
|
// write failed (connection error?) — fallback to RESP2 and perform AUTH if needed
|
|
resp2_auth(config.password, mut db)!
|
|
return db
|
|
}
|
|
|
|
// Try to read and consume HELLO response. If HELLO fails (unknown command / error),
|
|
// treat as not-supported and fall back to RESP2 + AUTH.
|
|
db.read_response() or {
|
|
resp2_auth(config.password, mut db)!
|
|
return db
|
|
}
|
|
|
|
return db
|
|
}
|
|
|
|
// close terminates the connection to Redis server
|
|
pub fn (mut db DB) close() ! {
|
|
if db.tls {
|
|
db.ssl_conn.close()!
|
|
} else {
|
|
db.conn.close()!
|
|
}
|
|
}
|
|
|
|
// Helper methods for TLS abstraction
|
|
fn (mut db DB) write_data(data []u8) ! {
|
|
if db.tls {
|
|
db.ssl_conn.write(data)!
|
|
} else {
|
|
db.conn.write(data)!
|
|
}
|
|
}
|
|
|
|
fn (mut db DB) read_data(mut buf []u8) !int {
|
|
if db.tls {
|
|
return db.ssl_conn.read(mut buf)
|
|
} else {
|
|
return db.conn.read(mut buf)
|
|
}
|
|
}
|
|
|
|
fn (mut db DB) read_ptr_data(ptr &u8, len int) !int {
|
|
if db.tls {
|
|
return db.ssl_conn.socket_read_into_ptr(ptr, len)!
|
|
} else {
|
|
return db.conn.read_ptr(ptr, len)!
|
|
}
|
|
}
|
|
|
|
// auth sends an AUTH command to the server with the given password.
|
|
pub fn (mut db DB) auth(password string) ! {
|
|
resp := db.cmd('AUTH', password)!
|
|
match resp {
|
|
string {
|
|
if resp != 'OK' {
|
|
return error('Authentication failed: ${resp}')
|
|
}
|
|
}
|
|
else {
|
|
return error('Authentication failed: unexpected response type')
|
|
}
|
|
}
|
|
}
|
|
|
|
// ping sends a PING command to verify server responsiveness
|
|
pub fn (mut db DB) ping() !string {
|
|
return db.cmd('PING')! as string
|
|
}
|
|
|
|
// validate checks whether the Redis connection is still responsive.
|
|
pub fn (mut db DB) validate() !bool {
|
|
return db.ping()! == 'PONG'
|
|
}
|
|
|
|
// reset clears any queued pipeline state before the connection is reused.
|
|
pub fn (mut db DB) reset() ! {
|
|
db.pipeline_mode = false
|
|
db.pipeline_cmd_count = 0
|
|
db.cmd_buf.clear()
|
|
db.resp_buf.clear()
|
|
db.pipeline_buffer.clear()
|
|
}
|
|
|
|
// del deletes a `key`
|
|
pub fn (mut db DB) del(key string) !i64 {
|
|
// *2\r\n$3\r\nDEL\r\n$6\r\ncounter\r\n
|
|
// send cmd
|
|
db.cmd_buf.clear()
|
|
db.cmd_buf << '*2\r\n$3\r\nDEL\r\n$${key.len}\r\n${key}\r\n'.bytes()
|
|
if db.pipeline_mode {
|
|
db.pipeline_buffer << db.cmd_buf
|
|
db.pipeline_cmd_count++
|
|
} else {
|
|
db.write_data(db.cmd_buf)!
|
|
|
|
// read resp
|
|
return db.read_response()! as i64
|
|
}
|
|
return 0
|
|
}
|
|
|
|
// set stores a `key`-value` pair in Redis. Supported value types: number, string, []u8
|
|
pub fn (mut db DB) set[T](key string, value T) !string {
|
|
// *3\r\n$3\r\nSET\r\n$4\r\nname\r\n$5\r\nVlang\r\n
|
|
db.cmd_buf.clear()
|
|
db.cmd_buf << '*3\r\n$3\r\nSET\r\n$${key.len}\r\n${key}\r\n'.bytes()
|
|
$if T is $int {
|
|
val_str := value.str()
|
|
db.cmd_buf << '$${val_str.len}\r\n${val_str}'.bytes()
|
|
} $else $if T is string {
|
|
db.cmd_buf << '$${value.len}\r\n${value}'.bytes()
|
|
} $else $if T is []u8 {
|
|
db.cmd_buf << '$${value.len}\r\n'.bytes()
|
|
db.cmd_buf << value
|
|
} $else {
|
|
return error('`set()`: unsupported value type. Allowed: number, string, []u8')
|
|
}
|
|
db.cmd_buf << '\r\n'.bytes()
|
|
if db.pipeline_mode {
|
|
db.pipeline_buffer << db.cmd_buf
|
|
db.pipeline_cmd_count++
|
|
} else {
|
|
db.write_data(db.cmd_buf)!
|
|
return db.read_response()! as string
|
|
}
|
|
return ''
|
|
}
|
|
|
|
// get retrieves the value of a `key`. Supported return types: string, int, []u8
|
|
pub fn (mut db DB) get[T](key string) !T {
|
|
// *2\r\n$3\r\nGET\r\n$4\r\nname\r\n
|
|
// send cmd
|
|
db.cmd_buf.clear()
|
|
db.cmd_buf << '*2\r\n$3\r\nGET\r\n$${key.len}\r\n${key}\r\n'.bytes()
|
|
if db.pipeline_mode {
|
|
db.pipeline_buffer << db.cmd_buf
|
|
db.pipeline_cmd_count++
|
|
} else {
|
|
db.write_data(db.cmd_buf)!
|
|
resp := db.read_response()!
|
|
match resp {
|
|
[]u8 {
|
|
$if T is string {
|
|
return resp.bytestr()
|
|
} $else $if T is $int {
|
|
return T(resp.bytestr().i64())
|
|
} $else $if T is []u8 {
|
|
return resp
|
|
}
|
|
}
|
|
RedisNull {
|
|
return error('`get()`: key ${key} not found')
|
|
}
|
|
else {
|
|
return error('`get()`: unexpected response type')
|
|
}
|
|
}
|
|
|
|
return error('`get()`: unsupported data type')
|
|
}
|
|
return T{}
|
|
}
|
|
|
|
// incr increments the integer value of a `key` by 1
|
|
pub fn (mut db DB) incr(key string) !i64 {
|
|
// *2\r\n$4\r\nINCR\r\n$6\r\ncounter\r\n
|
|
// send cmd
|
|
db.cmd_buf.clear()
|
|
db.cmd_buf << '*2\r\n$4\r\nINCR\r\n$${key.len}\r\n${key}\r\n'.bytes()
|
|
if db.pipeline_mode {
|
|
db.pipeline_buffer << db.cmd_buf
|
|
db.pipeline_cmd_count++
|
|
} else {
|
|
db.write_data(db.cmd_buf)!
|
|
|
|
// read resp
|
|
return db.read_response()! as i64
|
|
}
|
|
return 0
|
|
}
|
|
|
|
// decr decrements the integer value of a `key` by 1
|
|
pub fn (mut db DB) decr(key string) !i64 {
|
|
// *2\r\n$4\r\nDECR\r\n$6\r\ncounter\r\n
|
|
// send cmd
|
|
db.cmd_buf.clear()
|
|
db.cmd_buf << '*2\r\n$4\r\nDECR\r\n$${key.len}\r\n${key}\r\n'.bytes()
|
|
if db.pipeline_mode {
|
|
db.pipeline_buffer << db.cmd_buf
|
|
db.pipeline_cmd_count++
|
|
} else {
|
|
db.write_data(db.cmd_buf)!
|
|
|
|
// read resp
|
|
return db.read_response()! as i64
|
|
}
|
|
return 0
|
|
}
|
|
|
|
// hset sets multiple `key`-`value` pairs in a hash. Supported value types: string, int, []u8
|
|
pub fn (mut db DB) hset[T](key string, m map[string]T) !int {
|
|
// HSET user:1 name "John" age 30
|
|
// *6\r\n$4\r\nHSET\r\n$6\r\nuser:1\r\n$4\r\nname\r\n$4\r\nJohn\r\n$3\r\nage\r\n$2\r\n30\r\n
|
|
db.cmd_buf.clear()
|
|
db.cmd_buf << '*${2 + m.len * 2}\r\n$4\r\nHSET\r\n$${key.len}\r\n${key}\r\n'.bytes()
|
|
for k, v in m {
|
|
db.cmd_buf << '$${k.len}\r\n${k}\r\n'.bytes()
|
|
$if T is string {
|
|
db.cmd_buf << '$${v.len}\r\n${v}\r\n'.bytes()
|
|
} $else $if T is $int {
|
|
v_str := v.str()
|
|
db.cmd_buf << '$${v_str.len}\r\n${v_str}\r\n'.bytes()
|
|
} $else $if T is []u8 {
|
|
// Write bulk string header correctly (no stray '$' after the length)
|
|
db.cmd_buf << '$${v.len}\r\n'.bytes()
|
|
db.cmd_buf << v
|
|
db.cmd_buf << '\r\n'.bytes()
|
|
} $else {
|
|
return error('`hset()`: unsupported value type. Allowed: number, string, []u8')
|
|
}
|
|
}
|
|
if db.pipeline_mode {
|
|
db.pipeline_buffer << db.cmd_buf
|
|
db.pipeline_cmd_count++
|
|
} else {
|
|
db.write_data(db.cmd_buf)!
|
|
return int(db.read_response()! as i64)
|
|
}
|
|
return 0
|
|
}
|
|
|
|
// hget retrieves the value of a hash field. Supported return types: string, int, []u8
|
|
pub fn (mut db DB) hget[T](key string, m_key string) !T {
|
|
// HGET user:1 name
|
|
// *3\r\n$4\r\nHGET\r\n$6\r\nuser:1\r\n$4\r\nname\r\n
|
|
db.cmd_buf.clear()
|
|
db.cmd_buf << '*3\r\n$4\r\nHGET\r\n$${key.len}\r\n${key}\r\n'.bytes()
|
|
db.cmd_buf << '$${m_key.len}\r\n${m_key}\r\n'.bytes()
|
|
if db.pipeline_mode {
|
|
db.pipeline_buffer << db.cmd_buf
|
|
db.pipeline_cmd_count++
|
|
} else {
|
|
db.write_data(db.cmd_buf)!
|
|
resp := db.read_response()! as []u8
|
|
$if T is string {
|
|
return resp.bytestr()
|
|
} $else $if T is $int {
|
|
return resp.bytestr().i64()
|
|
} $else $if T is []u8 {
|
|
return resp
|
|
}
|
|
return error('`hget()`: unsupported return type. Allowed: number, string, []u8')
|
|
}
|
|
return T{}
|
|
}
|
|
|
|
// hgetall retrieves all fields and values of a hash. Supported value types: string, int, []u8
|
|
pub fn (mut db DB) hgetall[T](key string) !map[string]T {
|
|
// HGETALL user:1
|
|
// *2\r\n$7\r\nHGETALL\r\n$6\r\nuser:1\r\n
|
|
$if T !is string && T !is $int && T !is []u8 {
|
|
return error('`hgetall()`: unsupported value type. Allowed: number, string, []u8')
|
|
}
|
|
db.cmd_buf.clear()
|
|
db.cmd_buf << '*2\r\n$7\r\nHGETALL\r\n$${key.len}\r\n${key}\r\n'.bytes()
|
|
if db.pipeline_mode {
|
|
db.pipeline_buffer << db.cmd_buf
|
|
db.pipeline_cmd_count++
|
|
} else {
|
|
db.write_data(db.cmd_buf)!
|
|
resp := db.read_response()!
|
|
|
|
// normalize result into map[string]T regardless of RESP2 array, RESP3 map,
|
|
// or RedisMap interleaved pairs.
|
|
$if T is string {
|
|
mut result := map[string]T{}
|
|
match resp {
|
|
[]RedisValue {
|
|
if resp.len % 2 != 0 {
|
|
return error('`hgetall()`: invalid HGETALL response format')
|
|
}
|
|
for i in 0 .. resp.len / 2 {
|
|
// keys and values expected as bulk strings for RESP2
|
|
key_val := resp[2 * i]
|
|
val_val := resp[2 * i + 1]
|
|
// key
|
|
k := match key_val {
|
|
[]u8 { key_val.bytestr() }
|
|
string { key_val }
|
|
else { return error('`hgetall()`: unexpected key type: ${key_val.type_name()}') }
|
|
}
|
|
|
|
// value
|
|
v := match val_val {
|
|
[]u8 { val_val.bytestr() }
|
|
string { val_val }
|
|
i64 { val_val.str() }
|
|
else { return error('`hgetall()`: unexpected value type: ${val_val.type_name()}') }
|
|
}
|
|
|
|
result[k] = v
|
|
}
|
|
return result
|
|
}
|
|
map[string]RedisValue {
|
|
for k, v in resp {
|
|
val_str := match v {
|
|
[]u8 { v.bytestr() }
|
|
string { v }
|
|
i64 { v.str() }
|
|
else { return error('`hgetall()`: unexpected value type in map: ${v.type_name()}') }
|
|
}
|
|
|
|
result[k] = val_str
|
|
}
|
|
return result
|
|
}
|
|
RedisMap {
|
|
pairs := resp.pairs
|
|
if pairs.len % 2 != 0 {
|
|
return error('`hgetall()`: invalid RedisMap response format')
|
|
}
|
|
for i := 0; i < pairs.len; i += 2 {
|
|
key_val := pairs[i]
|
|
val_val := pairs[i + 1]
|
|
k := match key_val {
|
|
[]u8 { key_val.bytestr() }
|
|
string { key_val }
|
|
else { return error('`hgetall()`: unexpected key type in RedisMap: ${key_val.type_name()}') }
|
|
}
|
|
|
|
v := match val_val {
|
|
[]u8 { val_val.bytestr() }
|
|
string { val_val }
|
|
i64 { val_val.str() }
|
|
else { return error('`hgetall()`: unexpected value type in RedisMap: ${val_val.type_name()}') }
|
|
}
|
|
|
|
result[k] = v
|
|
}
|
|
return result
|
|
}
|
|
else {
|
|
return error('`hgetall()`: unsupported response type: ${resp.type_name()}')
|
|
}
|
|
}
|
|
} $else $if T is $int {
|
|
mut result := map[string]T{}
|
|
match resp {
|
|
[]RedisValue {
|
|
if resp.len % 2 != 0 {
|
|
return error('`hgetall()`: invalid HGETALL response format')
|
|
}
|
|
for i in 0 .. resp.len / 2 {
|
|
key_val := resp[2 * i]
|
|
val_val := resp[2 * i + 1]
|
|
k := match key_val {
|
|
[]u8 { key_val.bytestr() }
|
|
string { key_val }
|
|
else { return error('`hgetall()`: unexpected key type: ${key_val.type_name()}') }
|
|
}
|
|
|
|
v := match val_val {
|
|
[]u8 { val_val.bytestr().i64() }
|
|
string { val_val.i64() }
|
|
i64 { val_val }
|
|
else { return error('`hgetall()`: unexpected value type: ${val_val.type_name()}') }
|
|
}
|
|
|
|
result[k] = T(v)
|
|
}
|
|
return result
|
|
}
|
|
map[string]RedisValue {
|
|
for k, v in resp {
|
|
n := match v {
|
|
[]u8 { v.bytestr().i64() }
|
|
string { v.i64() }
|
|
i64 { v }
|
|
else { return error('`hgetall()`: unexpected value type in map: ${v.type_name()}') }
|
|
}
|
|
|
|
result[k] = T(n)
|
|
}
|
|
return result
|
|
}
|
|
RedisMap {
|
|
pairs := resp.pairs
|
|
if pairs.len % 2 != 0 {
|
|
return error('`hgetall()`: invalid RedisMap response format')
|
|
}
|
|
for i := 0; i < pairs.len; i += 2 {
|
|
key_val := pairs[i]
|
|
val_val := pairs[i + 1]
|
|
k := match key_val {
|
|
[]u8 { key_val.bytestr() }
|
|
string { key_val }
|
|
else { return error('`hgetall()`: unexpected key type in RedisMap: ${key_val.type_name()}') }
|
|
}
|
|
|
|
n := match val_val {
|
|
[]u8 { val_val.bytestr().i64() }
|
|
string { val_val.i64() }
|
|
i64 { val_val }
|
|
else { return error('`hgetall()`: unexpected value type in RedisMap: ${val_val.type_name()}') }
|
|
}
|
|
|
|
result[k] = T(n)
|
|
}
|
|
return result
|
|
}
|
|
else {
|
|
return error('`hgetall()`: unsupported response type: ${resp.type_name()}')
|
|
}
|
|
}
|
|
} $else $if T is []u8 {
|
|
mut result := map[string]T{}
|
|
match resp {
|
|
[]RedisValue {
|
|
if resp.len % 2 != 0 {
|
|
return error('`hgetall()`: invalid HGETALL response format')
|
|
}
|
|
for i in 0 .. resp.len / 2 {
|
|
key_val := resp[2 * i]
|
|
val_val := resp[2 * i + 1]
|
|
k := match key_val {
|
|
[]u8 { key_val.bytestr() }
|
|
string { key_val }
|
|
else { return error('`hgetall()`: unexpected key type: ${key_val.type_name()}') }
|
|
}
|
|
|
|
v := match val_val {
|
|
[]u8 { val_val }
|
|
string { val_val.bytes() }
|
|
i64 { val_val.str().bytes() }
|
|
else { return error('`hgetall()`: unexpected value type: ${val_val.type_name()}') }
|
|
}
|
|
|
|
result[k] = v
|
|
}
|
|
return result
|
|
}
|
|
map[string]RedisValue {
|
|
for k, v in resp {
|
|
b := match v {
|
|
[]u8 { v }
|
|
string { v.bytes() }
|
|
i64 { v.str().bytes() }
|
|
else { return error('`hgetall()`: unexpected value type in map: ${v.type_name()}') }
|
|
}
|
|
|
|
result[k] = b
|
|
}
|
|
return result
|
|
}
|
|
RedisMap {
|
|
pairs := resp.pairs
|
|
if pairs.len % 2 != 0 {
|
|
return error('`hgetall()`: invalid RedisMap response format')
|
|
}
|
|
for i := 0; i < pairs.len; i += 2 {
|
|
key_val := pairs[i]
|
|
val_val := pairs[i + 1]
|
|
k := match key_val {
|
|
[]u8 { key_val.bytestr() }
|
|
string { key_val }
|
|
else { return error('`hgetall()`: unexpected key type in RedisMap: ${key_val.type_name()}') }
|
|
}
|
|
|
|
b := match val_val {
|
|
[]u8 { val_val }
|
|
string { val_val.bytes() }
|
|
i64 { val_val.str().bytes() }
|
|
else { return error('`hgetall()`: unexpected value type in RedisMap: ${val_val.type_name()}') }
|
|
}
|
|
|
|
result[k] = b
|
|
}
|
|
return result
|
|
}
|
|
else {
|
|
return error('`hgetall()`: unsupported response type: ${resp.type_name()}')
|
|
}
|
|
}
|
|
} $else {
|
|
// should not happen due to compile-time check above
|
|
return error('`hgetall()`: unsupported value type ${T.type_name()}')
|
|
}
|
|
}
|
|
return map[string]T{}
|
|
}
|
|
|
|
// expire sets a `key`'s time to live in `seconds`
|
|
pub fn (mut db DB) expire(key string, seconds int) !bool {
|
|
// *3\r\n$6\r\nEXPIRE\r\n$6\r\ncounter\r\n$3\r\n600\r\n
|
|
// send cmd
|
|
seconds_str := seconds.str()
|
|
db.cmd_buf.clear()
|
|
db.cmd_buf << '*3\r\n$6\r\nEXPIRE\r\n$${key.len}\r\n${key}\r\n'.bytes()
|
|
db.cmd_buf << '$${seconds_str.len}\r\n${seconds_str}\r\n'.bytes()
|
|
if db.pipeline_mode {
|
|
db.pipeline_buffer << db.cmd_buf
|
|
db.pipeline_cmd_count++
|
|
} else {
|
|
db.write_data(db.cmd_buf)!
|
|
|
|
// read resp
|
|
rv := db.read_response()!
|
|
|
|
// normalize to boolean result as before
|
|
match rv {
|
|
i64 { return rv != 0 }
|
|
[]u8 { return rv.bytestr().i64() != 0 }
|
|
string { return rv.i64() != 0 }
|
|
else { return error('`expire()`: unexpected response type: ${rv.type_name()}') }
|
|
}
|
|
}
|
|
return false
|
|
}
|
|
|
|
// read_response_bulk_string handles Redis bulk string responses (format: $<length>\r\n<data>\r\n)
|
|
fn (mut db DB) read_response_bulk_string() !RedisValue {
|
|
mut data_length := i64(-1)
|
|
mut chunk := []u8{len: 1}
|
|
|
|
db.resp_buf.clear()
|
|
for {
|
|
bytes_read := db.read_data(mut chunk) or {
|
|
return error('`read_response_bulk_string()`: connection error ${err}')
|
|
}
|
|
if bytes_read == 0 {
|
|
return error('`read_response_bulk_string()`: connection closed prematurely')
|
|
}
|
|
db.resp_buf << chunk[0]
|
|
|
|
if chunk[0] == `\n` {
|
|
break
|
|
}
|
|
if (chunk[0] < `0` || chunk[0] > `9`) && chunk[0] != `\r` && chunk[0] != `-` {
|
|
return error('`read_response_bulk_string()`: invalid bulk string header')
|
|
}
|
|
}
|
|
|
|
if db.resp_buf.len < 2 {
|
|
return error('`read_response_bulk_string()`: bulk string header too short')
|
|
}
|
|
|
|
data_length = db.resp_buf[0..db.resp_buf.len - 2].bytestr().i64()
|
|
|
|
// -1 -> NULL bulk string
|
|
if data_length == -1 {
|
|
return RedisNull{}
|
|
}
|
|
|
|
// If zero-length payload, read exactly the 2-byte terminator CRLF reliably
|
|
if data_length == 0 {
|
|
mut term := [2]u8{}
|
|
n := db.read_ptr_data(&term[0], 2)! // read remaining terminator bytes
|
|
match n {
|
|
0, 1 {
|
|
return error('`read_response_bulk_string()`: incomplete terminator for empty string')
|
|
}
|
|
else {
|
|
if n > 2 {
|
|
return error('`read_response_bulk_string()`: should never get here - more than 2 bytes read?!?')
|
|
}
|
|
}
|
|
}
|
|
|
|
if term[0] != `\r` || term[1] != `\n` {
|
|
return error('invalid terminator for empty string')
|
|
}
|
|
return []u8{}
|
|
}
|
|
|
|
// Read payload of exactly data_length bytes
|
|
mut data_buf := []u8{len: int(data_length)}
|
|
mut total_read := 0
|
|
for total_read < data_buf.len {
|
|
mut ptr := unsafe { &data_buf[total_read] }
|
|
n := db.read_ptr_data(ptr, data_buf.len - total_read)!
|
|
if n == 0 && total_read < data_buf.len {
|
|
return error('`read_response_bulk_string()`: incomplete data: read ${total_read} / ${data_buf.len} bytes')
|
|
}
|
|
total_read += n
|
|
}
|
|
|
|
// Now read the trailing CRLF terminator (2 bytes) reliably
|
|
mut term := []u8{len: 2}
|
|
mut term_read := 0
|
|
for term_read < 2 {
|
|
mut ptr := unsafe { &term[term_read] }
|
|
n := db.read_ptr_data(ptr, 2 - term_read)!
|
|
if n == 0 && term_read < 2 {
|
|
return error('`read_response_bulk_string()`: incomplete terminator after payload')
|
|
}
|
|
term_read += n
|
|
}
|
|
if term[0] != `\r` || term[1] != `\n` {
|
|
return error('`read_response_bulk_string()`: invalid data terminator')
|
|
}
|
|
|
|
return data_buf
|
|
}
|
|
|
|
// read_header reads a CRLF-terminated header (returns content without trailing CRLF)
|
|
fn (mut db DB) read_header() !string {
|
|
mut chunk := []u8{len: 1}
|
|
db.resp_buf.clear()
|
|
for {
|
|
bytes_read := db.read_data(mut chunk) or {
|
|
return error('`read_header()`: connection error ${err}')
|
|
}
|
|
if bytes_read == 0 {
|
|
return error('`read_header()`: connection closed prematurely')
|
|
}
|
|
db.resp_buf << chunk[0]
|
|
if chunk[0] == `\n` {
|
|
break
|
|
}
|
|
}
|
|
if db.resp_buf.len < 2 {
|
|
return error('`read_header()`: header too short')
|
|
}
|
|
return db.resp_buf[0..db.resp_buf.len - 2].bytestr()
|
|
}
|
|
|
|
// read_exact_payload reads exactly n bytes + trailing CRLF and return the data bytes (without CRLF)
|
|
fn (mut db DB) read_exact_payload(n int) ![]u8 {
|
|
if n < 0 {
|
|
return error('invalid payload length ${n}')
|
|
}
|
|
mut data_buf := []u8{len: n + 2}
|
|
mut total_read := 0
|
|
for total_read < data_buf.len {
|
|
remaining := data_buf.len - total_read
|
|
chunk_size := if remaining > 1 { 1 } else { remaining }
|
|
mut chunk_ptr := unsafe { &data_buf[total_read] }
|
|
|
|
bytes_read := db.read_ptr_data(chunk_ptr, chunk_size)!
|
|
total_read += bytes_read
|
|
|
|
if bytes_read == 0 && total_read < data_buf.len {
|
|
return error('`read_exact_payload()`: incomplete data: read ${total_read} / ${data_buf.len} bytes')
|
|
}
|
|
}
|
|
// must ending with CRLF
|
|
if data_buf[n] != `\r` || data_buf[n + 1] != `\n` {
|
|
return error('`read_exact_payload()`: invalid data terminator')
|
|
}
|
|
return data_buf[0..n].clone()
|
|
}
|
|
|
|
// read_resp3_boolean_payload handles RESP3 boolean (#t or #f)
|
|
fn (mut db DB) read_resp3_boolean_payload() !bool {
|
|
s := db.read_header()!
|
|
if s == 't' {
|
|
return true
|
|
}
|
|
if s == 'f' {
|
|
return false
|
|
}
|
|
return error('`read_resp3_boolean_payload()`: invalid boolean: ${s}')
|
|
}
|
|
|
|
// read_resp3_double_payload handles RESP3 double (,<double>)
|
|
fn (mut db DB) read_resp3_double_payload() !f64 {
|
|
s := db.read_header()!
|
|
return s.f64()
|
|
}
|
|
|
|
// read_resp3_bignum_payload handles RESP3 big number ((<number>) -> big.Integer)
|
|
fn (mut db DB) read_resp3_bignum_payload() !big.Integer {
|
|
mut s := db.read_header()!
|
|
// RESP3 bignum frames may be wrapped in parentheses, e.g. "(12345)".
|
|
// Trim leading '(' and trailing ')' if present to make the numeric string safe for the parser.
|
|
if s.len > 0 && s[0] == `(` {
|
|
s = s[1..]
|
|
}
|
|
if s.len > 0 && s[s.len - 1] == `)` {
|
|
s = s[0..s.len - 1]
|
|
}
|
|
return big.integer_from_string(s)!
|
|
}
|
|
|
|
// read_resp3_blob_error_payload handles RESP3 blob error (!<len>\r\n<data>\r\n)
|
|
fn (mut db DB) read_resp3_blob_error_payload() !RedisBlobError {
|
|
header := db.read_header()!
|
|
length := header.i64()
|
|
if length == -1 {
|
|
return RedisBlobError{
|
|
data: []u8{}
|
|
}
|
|
}
|
|
payload := db.read_exact_payload(int(length))!
|
|
return RedisBlobError{
|
|
data: payload
|
|
}
|
|
}
|
|
|
|
// read_resp3_verbatim_payload handles RESP3 verbatim (=<len>\r\n<fmt>:<data>\r\n)
|
|
fn (mut db DB) read_resp3_verbatim_payload() !RedisVerbatim {
|
|
header := db.read_header()!
|
|
length := header.i64()
|
|
if length == -1 {
|
|
return RedisVerbatim{
|
|
format: ''
|
|
data: []u8{}
|
|
}
|
|
}
|
|
payload := db.read_exact_payload(int(length))!
|
|
// split at first ':'
|
|
idx := payload.bytestr().index(':') or { -1 }
|
|
if idx == -1 {
|
|
return RedisVerbatim{
|
|
format: ''
|
|
data: payload
|
|
}
|
|
}
|
|
fmt := payload[0..idx].bytestr()
|
|
data := payload[idx + 1..].clone()
|
|
return RedisVerbatim{
|
|
format: fmt
|
|
data: data
|
|
}
|
|
}
|
|
|
|
// read_resp3_map_payload handles RESP3 map (%) where header is number of key/value pairs
|
|
// Try to return map[string]RedisValue when keys are string-like, otherwise return RedisMap
|
|
fn (mut db DB) read_resp3_map_payload() !RedisValue {
|
|
header := db.read_header()!
|
|
count := header.i64()
|
|
if count == -1 {
|
|
return RedisNull{}
|
|
}
|
|
if count == 0 {
|
|
return map[string]RedisValue{}
|
|
}
|
|
mut pairs := []RedisValue{cap: int(count) * 2}
|
|
for _ in 0 .. count {
|
|
key := db.read_response()!
|
|
val := db.read_response()!
|
|
pairs << key
|
|
pairs << val
|
|
}
|
|
// attempt to convert to map[string]RedisValue
|
|
mut kv := map[string]RedisValue{}
|
|
for i := 0; i < pairs.len; i += 2 {
|
|
k := pairs[i]
|
|
v := pairs[i + 1]
|
|
match k {
|
|
[]u8 {
|
|
kv[k.bytestr()] = v
|
|
}
|
|
string {
|
|
kv[k] = v
|
|
}
|
|
else {
|
|
// fallback: return interleaved pairs preserved as RedisMap
|
|
return RedisMap{
|
|
pairs: pairs
|
|
}
|
|
}
|
|
}
|
|
}
|
|
return kv
|
|
}
|
|
|
|
// read_resp3_attr_payload handles RESP3 attributes/attrs (|) and returns a map[string]RedisValue
|
|
// Attributes are map-like and we return a map when keys are string-like. If a non-string
|
|
// key is encountered, this treats it as an error (attributes are expected to be string-keyed).
|
|
fn (mut db DB) read_resp3_attr_payload() !RedisValue {
|
|
header := db.read_header()!
|
|
count := header.i64()
|
|
if count == -1 {
|
|
return RedisNull{}
|
|
}
|
|
if count == 0 {
|
|
return map[string]RedisValue{}
|
|
}
|
|
mut kv := map[string]RedisValue{}
|
|
for _ in 0 .. count {
|
|
k := db.read_response()!
|
|
v := db.read_response()!
|
|
match k {
|
|
[]u8 {
|
|
kv[k.bytestr()] = v
|
|
}
|
|
string {
|
|
kv[k] = v
|
|
}
|
|
else {
|
|
return error('`read_resp3_attr_payload()`: attribute key is not a string-like type')
|
|
}
|
|
}
|
|
}
|
|
return kv
|
|
}
|
|
|
|
// read_resp3_set_payload handles RESP3 set (~)
|
|
fn (mut db DB) read_resp3_set_payload() !RedisSet {
|
|
header := db.read_header()!
|
|
count := header.i64()
|
|
if count == -1 {
|
|
return RedisSet{
|
|
elements: []RedisValue{}
|
|
}
|
|
}
|
|
mut elems := []RedisValue{cap: int(count)}
|
|
for _ in 0 .. count {
|
|
elems << db.read_response()!
|
|
}
|
|
return RedisSet{
|
|
elements: elems
|
|
}
|
|
}
|
|
|
|
// read_resp3_push_payload handles RESP3 push (>) - array-like
|
|
fn (mut db DB) read_resp3_push_payload() !RedisPush {
|
|
header := db.read_header()!
|
|
count := header.i64()
|
|
if count == -1 {
|
|
return RedisPush{
|
|
elements: []RedisValue{}
|
|
}
|
|
}
|
|
mut elems := []RedisValue{cap: int(count)}
|
|
for _ in 0 .. count {
|
|
elems << db.read_response()!
|
|
}
|
|
return RedisPush{
|
|
elements: elems
|
|
}
|
|
}
|
|
|
|
// read_response_i64 handles Redis integer responses (format: :<number>\r\n)
|
|
fn (mut db DB) read_response_i64() !i64 {
|
|
db.resp_buf.clear()
|
|
unsafe { db.resp_buf.grow_len(resp_buf_pre_allocate_len) }
|
|
mut total_read := 0
|
|
|
|
for total_read < db.resp_buf.len {
|
|
remaining := db.resp_buf.len - total_read
|
|
chunk_size := if remaining > 1 { 1 } else { remaining }
|
|
mut chunk_ptr := unsafe { &db.resp_buf[total_read] }
|
|
|
|
bytes_read := db.read_ptr_data(chunk_ptr, chunk_size)!
|
|
total_read += bytes_read
|
|
|
|
if total_read > 2 {
|
|
if db.resp_buf[total_read - 2] == `\r` && db.resp_buf[total_read - 1] == `\n` {
|
|
break
|
|
}
|
|
}
|
|
if bytes_read == 0 {
|
|
return error('`read_response_i64()`: incomplete data: read ${total_read} bytes')
|
|
}
|
|
}
|
|
ret_val := db.resp_buf[0..total_read - 2].bytestr().i64()
|
|
return ret_val
|
|
}
|
|
|
|
// read_response_simple_string handles Redis simple string responses (format: +<string>\r\n)
|
|
fn (mut db DB) read_response_simple_string() !string {
|
|
db.resp_buf.clear()
|
|
unsafe { db.resp_buf.grow_len(resp_buf_pre_allocate_len) }
|
|
mut total_read := 0
|
|
|
|
for total_read < db.resp_buf.len {
|
|
remaining := db.resp_buf.len - total_read
|
|
chunk_size := if remaining > 1 { 1 } else { remaining }
|
|
mut chunk_ptr := unsafe { &db.resp_buf[total_read] }
|
|
|
|
bytes_read := db.read_ptr_data(chunk_ptr, chunk_size)!
|
|
total_read += bytes_read
|
|
|
|
if total_read > 2 {
|
|
if db.resp_buf[total_read - 2] == `\r` && db.resp_buf[total_read - 1] == `\n` {
|
|
break
|
|
}
|
|
}
|
|
if bytes_read == 0 {
|
|
return error('`read_response_simple_string()`: incomplete data: read ${total_read} bytes')
|
|
}
|
|
}
|
|
return db.resp_buf[0..total_read - 2].bytestr()
|
|
}
|
|
|
|
// read_response_array handles Redis array responses (format: *<length>\r\n<elements>)
|
|
fn (mut db DB) read_response_array() !RedisValue {
|
|
mut array_len := i64(-1)
|
|
mut chunk := []u8{len: 1}
|
|
|
|
db.resp_buf.clear()
|
|
for {
|
|
bytes_read := db.read_data(mut chunk) or {
|
|
return error('`read_response_array()`: connection error: ${err}')
|
|
}
|
|
if bytes_read == 0 {
|
|
return error('`read_response_array()`: connection closed prematurely')
|
|
}
|
|
db.resp_buf << chunk[0]
|
|
|
|
if chunk[0] == `\n` {
|
|
break
|
|
}
|
|
if (chunk[0] < `0` || chunk[0] > `9`) && chunk[0] != `\r` && chunk[0] != `-` {
|
|
return error('`read_response_array()`: invalid array header')
|
|
}
|
|
}
|
|
|
|
if db.resp_buf.len < 2 {
|
|
return error('`read_response_array()`: array header too short')
|
|
}
|
|
|
|
array_len = db.resp_buf[0..db.resp_buf.len - 2].bytestr().i64() // 排除\r\n
|
|
|
|
if array_len == -1 {
|
|
return RedisNull{}
|
|
}
|
|
if array_len == 0 {
|
|
return []RedisValue{}
|
|
}
|
|
|
|
mut elements := []RedisValue{cap: int(array_len)}
|
|
for _ in 0 .. array_len {
|
|
element := db.read_response() or {
|
|
return error('`read_response_array()`: failed to read array element: ${err}')
|
|
}
|
|
elements << element
|
|
}
|
|
return elements
|
|
}
|
|
|
|
// read_response handles all types of Redis responses (RESP2 + RESP3 when enabled)
|
|
fn (mut db DB) read_response() !RedisValue {
|
|
db.resp_buf.clear()
|
|
unsafe { db.resp_buf.grow_len(1) }
|
|
// Read the first non-empty, non-CR/LF prefix byte. Some transports or
|
|
// intermediate proxies may emit stray CR/LF bytes; skip them so we parse
|
|
// the actual RESP prefix correctly.
|
|
for {
|
|
read_len := db.read_data(mut db.resp_buf)!
|
|
if read_len != 1 {
|
|
return error('`read_response()`: empty response from server')
|
|
}
|
|
// Skip stray CR and LF bytes that may precede the real response prefix.
|
|
if db.resp_buf[0] == `\r` || db.resp_buf[0] == `\n` {
|
|
continue
|
|
}
|
|
break
|
|
}
|
|
|
|
// If the first non-CRLF byte is not a valid RESP prefix, attempt a bounded
|
|
// resynchronization: read and discard up to `max_skip` bytes looking for a
|
|
// valid prefix. This helps tolerate transient stray bytes while avoiding
|
|
// silently swallowing large amounts of data.
|
|
mut attempts := 0
|
|
|
|
for {
|
|
// If this byte is a known RESP prefix, proceed to parse normally.
|
|
ch := db.resp_buf[0]
|
|
if ch == `+` || ch == `-` || ch == `:` || ch == `$` || ch == `*` || ch == `#` || ch == `,`
|
|
|| ch == `(` || ch == `!` || ch == `=` || ch == `%` || ch == `~` || ch == `>`
|
|
|| ch == `|` {
|
|
break
|
|
}
|
|
// Give up after bounded attempts and return diagnostics.
|
|
if attempts >= max_skip {
|
|
mut prefix_val := -1
|
|
if db.resp_buf.len > 0 {
|
|
prefix_val = int(db.resp_buf[0])
|
|
}
|
|
mut hex := ''
|
|
for i in 0 .. db.resp_buf.len {
|
|
hex += '${int(db.resp_buf[i]):02x} '
|
|
}
|
|
return error("`read_response()`: unknown response prefix byte=${prefix_val} data_hex=\"${hex}\" data_str=\"${db.resp_buf.bytestr()}\"")
|
|
}
|
|
// Read and discard one more byte from the socket and treat it as the new candidate.
|
|
mut tmp := []u8{len: 1}
|
|
n := db.read_data(mut tmp) or { 0 }
|
|
if n == 0 {
|
|
return error('`read_response()`: incomplete data during resynchronization')
|
|
}
|
|
db.resp_buf[0] = tmp[0]
|
|
// skip CRLF if encountered but still count the attempt (we consumed a byte)
|
|
attempts++
|
|
continue
|
|
}
|
|
|
|
match db.resp_buf[0] {
|
|
`+` { // Simple string
|
|
return db.read_response_simple_string()!
|
|
}
|
|
`-` { // Error message
|
|
msg := db.read_response_simple_string()!
|
|
return error(msg)
|
|
}
|
|
`:` { // Integer
|
|
return db.read_response_i64()!
|
|
}
|
|
`$` { // Bulk string
|
|
return db.read_response_bulk_string()!
|
|
}
|
|
`*` { // Array
|
|
return db.read_response_array()!
|
|
}
|
|
// RESP3-only frames (enabled when db.version >= 3)
|
|
`#` { // Boolean
|
|
if db.version < 3 {
|
|
return error('`read_response()`: unknown response prefix: ${db.resp_buf.bytestr()}')
|
|
}
|
|
return RedisValue(db.read_resp3_boolean_payload()!)
|
|
}
|
|
`,` { // Double
|
|
if db.version < 3 {
|
|
return error('`read_response()`: unknown response prefix: ${db.resp_buf.bytestr()}')
|
|
}
|
|
return RedisValue(db.read_resp3_double_payload()!)
|
|
}
|
|
`(` { // Big number
|
|
if db.version < 3 {
|
|
return error('`read_response()`: unknown response prefix: ${db.resp_buf.bytestr()}')
|
|
}
|
|
return RedisValue(db.read_resp3_bignum_payload()!)
|
|
}
|
|
`!` { // Blob error
|
|
if db.version < 3 {
|
|
return error('`read_response()`: unknown response prefix: ${db.resp_buf.bytestr()}')
|
|
}
|
|
return db.read_resp3_blob_error_payload()!
|
|
}
|
|
`=` { // Verbatim string
|
|
if db.version < 3 {
|
|
return error('`read_response()`: unknown response prefix: ${db.resp_buf.bytestr()}')
|
|
}
|
|
return db.read_resp3_verbatim_payload()!
|
|
}
|
|
`%` { // Map
|
|
if db.version < 3 {
|
|
return error('`read_response()`: unknown response prefix: ${db.resp_buf.bytestr()}')
|
|
}
|
|
return db.read_resp3_map_payload()!
|
|
}
|
|
`~` { // Set
|
|
if db.version < 3 {
|
|
return error('`read_response()`: unknown response prefix: ${db.resp_buf.bytestr()}')
|
|
}
|
|
return RedisValue(db.read_resp3_set_payload()!)
|
|
}
|
|
`>` { // Push
|
|
if db.version < 3 {
|
|
return error('`read_response()`: unknown response prefix: ${db.resp_buf.bytestr()}')
|
|
}
|
|
return RedisValue(db.read_resp3_push_payload()!)
|
|
}
|
|
`|` { // Attr (map-like)
|
|
if db.version < 3 {
|
|
return error('`read_response()`: unknown response prefix: ${db.resp_buf.bytestr()}')
|
|
}
|
|
// Attributes are parsed like maps; reuse map parsing (attrs preserved as map[string]RedisValue)
|
|
return db.read_resp3_map_payload()!
|
|
}
|
|
else {
|
|
// Fallback: this should be unreachable because we validated prefixes above,
|
|
// but return a helpful diagnostic if it happens.
|
|
mut prefix_val := -1
|
|
if db.resp_buf.len > 0 {
|
|
prefix_val = int(db.resp_buf[0])
|
|
}
|
|
mut hex := ''
|
|
for i in 0 .. db.resp_buf.len {
|
|
hex += '${int(db.resp_buf[i]):02x} '
|
|
}
|
|
return error("`read_response()`: unknown response prefix byte=${prefix_val} data_hex=\"${hex}\" data_str=\"${db.resp_buf.bytestr()}\"")
|
|
}
|
|
}
|
|
|
|
return error('`read_response()`: unreachable code')
|
|
}
|
|
|
|
// cmd sends a custom command to Redis server
|
|
// for example: db.cmd('SET', 'key', 'value')!
|
|
pub fn (mut db DB) cmd(cmd ...string) !RedisValue {
|
|
mut sb := strings.new_builder(cmd.len * 20)
|
|
sb.write_string('*${cmd.len}\r\n') // Command array header
|
|
for arg in cmd {
|
|
sb.write_string('$${arg.len}\r\n${arg}\r\n')
|
|
}
|
|
if db.pipeline_mode {
|
|
db.pipeline_buffer << unsafe { sb.reuse_as_plain_u8_array() }
|
|
db.pipeline_cmd_count++
|
|
} else {
|
|
db.write_data(unsafe { sb.reuse_as_plain_u8_array() })!
|
|
return db.read_response()!
|
|
}
|
|
return RedisNull{}
|
|
}
|
|
|
|
// pipeline_start start a new pipeline
|
|
pub fn (mut db DB) pipeline_start() {
|
|
db.pipeline_mode = true
|
|
db.pipeline_cmd_count = 0
|
|
db.pipeline_buffer.clear()
|
|
}
|
|
|
|
// pipeline_execute executes the cmds in pipeline at once and retrieves all responses
|
|
pub fn (mut db DB) pipeline_execute() ![]RedisValue {
|
|
if !db.pipeline_mode {
|
|
return error('`pipeline_execute()`: pipeline not started')
|
|
}
|
|
defer {
|
|
db.pipeline_mode = false
|
|
}
|
|
if db.pipeline_buffer.len == 0 {
|
|
return []RedisValue{}
|
|
}
|
|
|
|
db.write_data(db.pipeline_buffer)!
|
|
|
|
mut results := []RedisValue{cap: db.pipeline_cmd_count}
|
|
for _ in 0 .. db.pipeline_cmd_count {
|
|
results << db.read_response()!
|
|
}
|
|
|
|
// reset to non-pipeline mode
|
|
db.pipeline_mode = false
|
|
db.pipeline_cmd_count = 0
|
|
return results
|
|
}
|