mtcontrib/mtsatellite
1
0
Fork 0
mirror of https://bitbucket.org/s_l_teichmann/mtsatellite synced 2024-12-23 16:50:18 +01:00
Code Issues Projects Releases Wiki Activity

unexport mtredisalize connection

Browse Source
This commit is contained in:
Sascha L. Teichmann 2024-01-07 04:22:22 +01:00
parent e73ccaf3a7
commit 447b8fc915
2 changed files with 22 additions and 22 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

42
cmd/mtredisalize/connection.go
View File

@ -23,22 +23,22 @@ var (
redisFalse = []byte(":0\r\n") redisFalse = []byte(":0\r\n")
) )
type Connection struct { type connection struct {
conn net.Conn conn net.Conn
session Session session Session
maxBulkStringSize int64 maxBulkStringSize int64
boolArray []bool boolArray []bool
} }
func NewConnection(conn net.Conn, session Session, maxBulkStringSize int64) *Connection { func newConnection(conn net.Conn, session Session, maxBulkStringSize int64) *connection {
return &Connection{ return &connection{
conn: conn, conn: conn,
session: session, session: session,
maxBulkStringSize: maxBulkStringSize, maxBulkStringSize: maxBulkStringSize,
boolArray: []bool{}} boolArray: []bool{}}
} }
func (c *Connection) Run() { func (c *connection) run() {
defer func() { defer func() {
c.session.Close() c.session.Close()
c.conn.Close() c.conn.Close()
@ -57,7 +57,7 @@ func logError(err error) bool {
return true return true
} }
func (c *Connection) Hdel(hash, key []byte) bool { func (c *connection) Hdel(hash, key []byte) bool {
success, err := c.session.Del(hash, key) success, err := c.session.Del(hash, key)
if err != nil { if err != nil {
@ -67,7 +67,7 @@ func (c *Connection) Hdel(hash, key []byte) bool {
return c.writeBool(success) return c.writeBool(success)
} }
func (c *Connection) Hget(hash, key []byte) bool { func (c *connection) Hget(hash, key []byte) bool {
var err error var err error
var data []byte var data []byte
@ -78,7 +78,7 @@ func (c *Connection) Hget(hash, key []byte) bool {
return c.writeBlock(data) return c.writeBlock(data)
} }
func (c *Connection) Hset(hash, key, data []byte) bool { func (c *connection) Hset(hash, key, data []byte) bool {
var err error var err error
var exists bool var exists bool
@ -94,7 +94,7 @@ func (c *Connection) Hset(hash, key, data []byte) bool {
return c.writeBool(exists) return c.writeBool(exists)
} }
func (c *Connection) Multi() bool { func (c *connection) Multi() bool {
if c.session.InTransaction() { if c.session.InTransaction() {
log.Println("WARN: Already running transaction.") log.Println("WARN: Already running transaction.")
} else { } else {
@ -105,7 +105,7 @@ func (c *Connection) Multi() bool {
return c.writeOk() return c.writeOk()
} }
func (c *Connection) Exec() bool { func (c *connection) Exec() bool {
if !c.session.InTransaction() { if !c.session.InTransaction() {
return c.writeEmptyArray() return c.writeEmptyArray()
} }
@ -117,7 +117,7 @@ func (c *Connection) Exec() bool {
return c.writeBoolArray(arr) return c.writeBoolArray(arr)
} }
func (c *Connection) Hkeys(hash []byte) bool { func (c *connection) Hkeys(hash []byte) bool {
var ( var (
err error err error
n int n int
@ -146,11 +146,11 @@ func (c *Connection) Hkeys(hash []byte) bool {
return true return true
} }
func (c *Connection) Ping() bool { func (c *connection) Ping() bool {
return c.writeMessage(redisPong) return c.writeMessage(redisPong)
} }
func (c *Connection) HSpatial(hash, first, second []byte) bool { func (c *connection) HSpatial(hash, first, second []byte) bool {
var ( var (
err error err error
blocks <-chan Block blocks <-chan Block
@ -174,16 +174,16 @@ func (c *Connection) HSpatial(hash, first, second []byte) bool {
return logError(c.writeBulkString(nil)) return logError(c.writeBulkString(nil))
} }
func (c *Connection) writeError(err error) bool { func (c *connection) writeError(err error) bool {
logError(err) logError(err)
return c.writeMessage(redisError) return c.writeMessage(redisError)
} }
func (c *Connection) writeEmptyArray() bool { func (c *connection) writeEmptyArray() bool {
return c.writeMessage(redisEmptyArray) return c.writeMessage(redisEmptyArray)
} }
func (c *Connection) writeBool(b bool) bool { func (c *connection) writeBool(b bool) bool {
if b { if b {
return c.writeMessage(redisTrue) return c.writeMessage(redisTrue)
} }
@ -195,7 +195,7 @@ func redisLength(prefix byte, s int) []byte {
return append(strconv.AppendInt(buf, int64(s), 10), '\r', '\n') return append(strconv.AppendInt(buf, int64(s), 10), '\r', '\n')
} }
func (c *Connection) writeBoolArray(arr []bool) bool { func (c *connection) writeBoolArray(arr []bool) bool {
if _, err := c.conn.Write(redisLength('*', len(arr))); err != nil { if _, err := c.conn.Write(redisLength('*', len(arr))); err != nil {
return logError(err) return logError(err)
} }
@ -207,24 +207,24 @@ func (c *Connection) writeBoolArray(arr []bool) bool {
return true return true
} }
func (c *Connection) writeMessage(msg []byte) bool { func (c *connection) writeMessage(msg []byte) bool {
_, err := c.conn.Write(msg) _, err := c.conn.Write(msg)
return logError(err) return logError(err)
} }
func (c *Connection) writeOk() bool { func (c *connection) writeOk() bool {
return c.writeMessage(redisOk) return c.writeMessage(redisOk)
} }
func (c *Connection) writeQueued() bool { func (c *connection) writeQueued() bool {
return c.writeMessage(redisQueued) return c.writeMessage(redisQueued)
} }
func (c *Connection) writeBlock(data []byte) bool { func (c *connection) writeBlock(data []byte) bool {
return logError(c.writeBulkString(data)) return logError(c.writeBulkString(data))
} }
func (c *Connection) writeBulkString(data []byte) (err error) { func (c *connection) writeBulkString(data []byte) (err error) {
con := c.conn con := c.conn
if data == nil { if data == nil {
_, err = con.Write(redisNoSuchBlock) _, err = con.Write(redisNoSuchBlock)

2
cmd/mtredisalize/main.go
View File

@ -160,7 +160,7 @@ func main() {
log.Printf("Cannot create session: %s\n", err) log.Printf("Cannot create session: %s\n", err)
conn.Close() conn.Close()
} else { } else {
go NewConnection(conn, session, maxBulkStringSize).Run() go newConnection(conn, session, maxBulkStringSize).run()
} }
case <-sigChan: case <-sigChan:
log.Println("Shutting down") log.Println("Shutting down")