result.go

package mock

import (
	"bufio"
	"bytes"
	"fmt"
	"net"
	"strconv"
	"sync"
	"sync/atomic"
	"time"
	"unsafe"

	"github.com/redis/rueidis"
)

func Result(val rueidis.RedisMessage) rueidis.RedisResult {
	r := result{val: val}
	return *(*rueidis.RedisResult)(unsafe.Pointer(&r))
}

func ErrorResult(err error) rueidis.RedisResult {
	r := result{err: err}
	return *(*rueidis.RedisResult)(unsafe.Pointer(&r))
}

func RedisString(v string) rueidis.RedisMessage {
	m := message{typ: '+', string: v}
	return *(*rueidis.RedisMessage)(unsafe.Pointer(&m))
}

func RedisBlobString(v string) rueidis.RedisMessage {
	m := message{typ: '$', string: v}
	return *(*rueidis.RedisMessage)(unsafe.Pointer(&m))
}

func RedisError(v string) rueidis.RedisMessage {
	m := message{typ: '-', string: v}
	return *(*rueidis.RedisMessage)(unsafe.Pointer(&m))
}

func RedisInt64(v int64) rueidis.RedisMessage {
	m := message{typ: ':', integer: v}
	return *(*rueidis.RedisMessage)(unsafe.Pointer(&m))
}

func RedisFloat64(v float64) rueidis.RedisMessage {
	m := message{typ: ',', string: strconv.FormatFloat(v, 'f', -1, 64)}
	return *(*rueidis.RedisMessage)(unsafe.Pointer(&m))
}

func RedisBool(v bool) rueidis.RedisMessage {
	m := message{typ: '#'}
	if v {
		m.integer = 1
	}
	return *(*rueidis.RedisMessage)(unsafe.Pointer(&m))
}

func RedisNil() rueidis.RedisMessage {
	m := message{typ: '_'}
	return *(*rueidis.RedisMessage)(unsafe.Pointer(&m))
}

func RedisArray(values ...rueidis.RedisMessage) rueidis.RedisMessage {
	m := message{typ: '*', values: values}
	return *(*rueidis.RedisMessage)(unsafe.Pointer(&m))
}

func RedisMap(kv map[string]rueidis.RedisMessage) rueidis.RedisMessage {
	values := make([]rueidis.RedisMessage, 0, 2*len(kv))
	for k, v := range kv {
		values = append(values, RedisString(k))
		values = append(values, v)
	}
	m := message{typ: '%', values: values}
	return *(*rueidis.RedisMessage)(unsafe.Pointer(&m))
}

func serialize(m message, buf *bytes.Buffer) {
	switch m.typ {
	case '$', '!', '=':
		buf.WriteString(fmt.Sprintf("%s%d\r\n%s\r\n", string(m.typ), len(m.string), m.string))
	case '+', '-', ',', '(':
		buf.WriteString(fmt.Sprintf("%s%s\r\n", string(m.typ), m.string))
	case ':', '#':
		buf.WriteString(fmt.Sprintf("%s%d\r\n", string(m.typ), m.integer))
	case '_':
		buf.WriteString(fmt.Sprintf("%s\r\n", string(m.typ)))
	case '*':
		buf.WriteString(fmt.Sprintf("%s%d\r\n", string(m.typ), len(m.values)))
		for _, v := range m.values {
			pv := *(*message)(unsafe.Pointer(&v))
			serialize(pv, buf)
		}
	case '%':
		buf.WriteString(fmt.Sprintf("%s%d\r\n", string(m.typ), len(m.values)/2))
		for _, v := range m.values {
			pv := *(*message)(unsafe.Pointer(&v))
			serialize(pv, buf)
		}
	}
}

func RedisResultStreamError(err error) rueidis.RedisResultStream {
	s := stream{e: err}
	return *(*rueidis.RedisResultStream)(unsafe.Pointer(&s))
}

func RedisResultStream(ms ...rueidis.RedisMessage) rueidis.RedisResultStream {
	buf := bytes.NewBuffer(nil)
	for _, m := range ms {
		pm := *(*message)(unsafe.Pointer(&m))
		serialize(pm, buf)
	}
	s := stream{n: len(ms), p: &pool{size: 1, cond: sync.NewCond(&sync.Mutex{})}, w: &pipe{r: bufio.NewReader(buf)}}
	return *(*rueidis.RedisResultStream)(unsafe.Pointer(&s))
}

func MultiRedisResultStream(ms ...rueidis.RedisMessage) rueidis.MultiRedisResultStream {
	return RedisResultStream(ms...)
}

func MultiRedisResultStreamError(err error) rueidis.RedisResultStream {
	return RedisResultStreamError(err)
}

type message struct {
	attrs   *rueidis.RedisMessage
	string  string
	values  []rueidis.RedisMessage
	integer int64
	typ     byte
	ttl     [7]byte
}

type result struct {
	err error
	val rueidis.RedisMessage
}

type pool struct {
	dead any
	cond *sync.Cond
	make func() any
	list []any
	size int
	down bool
}

type pipe struct {
	conn            net.Conn
	error           atomic.Value
	clhks           atomic.Value // closed hook, invoked after the conn is closed
	pshks           atomic.Value // pubsub hook, registered by the SetPubSubHooks
	queue           any
	cache           rueidis.CacheStore
	r               *bufio.Reader
	w               *bufio.Writer
	close           chan struct{}
	onInvalidations func([]rueidis.RedisMessage)
	r2psFn          func() (p *pipe, err error) // func to build pipe for resp2 pubsub
	r2pipe          *pipe                       // internal pipe for resp2 pubsub only
	ssubs           *any                        // pubsub smessage subscriptions
	nsubs           *any                        // pubsub  message subscriptions
	psubs           *any                        // pubsub pmessage subscriptions
	info            map[string]rueidis.RedisMessage
	timeout         time.Duration
	pinggap         time.Duration
	maxFlushDelay   time.Duration
	once            sync.Once
	r2mu            sync.Mutex
	version         int32
	_               [10]int32
	blcksig         int32
	state           int32
	waits           int32
	recvs           int32
	r2ps            bool // identify this pipe is used for resp2 pubsub or not
}

type stream struct {
	p *pool
	w *pipe
	e error
	n int
}