more updates

connections.go

@@ -86,34 +86,60 @@ var argsColor = color.New(color.FgGreen).SprintFunc()
 
 func (c *cmdLoggingHook) ProcessHook(hook redis.ProcessHook) redis.ProcessHook {
 	return func(ctx context.Context, cmd redis.Cmder) error {
-		var key string
-		var args []string
-
-		cmdArgs := cmd.Args()
-		name := cmdColor(strings.ToUpper(cmd.Name()))
-
-		if len(cmdArgs) > 1 {
-			key = keyColor(cmdArgs[1].(string))
-		}
-
-		if len(cmdArgs) > 2 {
-			for _, cmdArg := range cmdArgs[2:] {
-				args = append(args, argsColor(fmt.Sprintf("%v", cmdArg)))
-			}
-		}
+		name, key, args := cmdLogParts(cmd)
 
 		start := time.Now()
 		err := hook(ctx, cmd)
 		dur := float64(time.Since(start).Microseconds()) / 1000.0
 
-		fmt.Printf("Kredis (%.1fms) %s %s %s\n", dur, name, key, strings.Join(args, " "))
+		fmt.Printf("Kredis (%.1fms) %s %s %s\n", dur, name, key, args)
 
 		return err
 	}
 }
 
 func (c *cmdLoggingHook) ProcessPipelineHook(hook redis.ProcessPipelineHook) redis.ProcessPipelineHook {
 	return func(ctx context.Context, cmds []redis.Cmder) error {
-		return hook(ctx, cmds)
+
+		start := time.Now()
+		err := hook(ctx, cmds)
+		dur := float64(time.Since(start).Microseconds()) / 1000.0
+
+		for idx, cmd := range cmds {
+			name, key, args := cmdLogParts(cmd)
+
+			if idx == len(cmds)-1 {
+				fmt.Printf("Kredis (%.1fms) %s %s %s\n", dur, name, key, args)
+			} else {
+				fmt.Printf("Kredis (tx)    %s %s %s\n", name, key, args)
+			}
+		}
+
+		return err
 	}
 }
+
+func cmdLogParts(cmd redis.Cmder) (string, string, string) {
+	var key string
+	var args []string
+
+	cmdArgs := cmd.Args()
+	name := cmdColor(strings.ToUpper(cmd.Name()))
+
+	if len(cmdArgs) > 1 {
+		switch cmdArgs[1].(type) {
+		case int64:
+			key = keyColor(cmdArgs[1].(int64))
+		default:
+			key = keyColor(cmdArgs[1].(string))
+		}
+	}
+
+	if len(cmdArgs) > 2 {
+		for _, cmdArg := range cmdArgs[2:] {
+			args = append(args, argsColor(fmt.Sprintf("%v", cmdArg)))
+		}
+	}
+
+	return name, key, strings.Join(args, " ")
+}

iter.go

@@ -0,0 +1,67 @@
+package kredis
+
+// Loosely based on https://github.com/polyfloyd/go-iterator
+type iterator[T KredisTyped] interface {
+	next() (T, bool)
+	values() []interface{}
+	unique() []interface{}
+	uniqueMap() map[T]struct{}
+}
+
+type iter[T KredisTyped] struct {
+	elements []T // rename "elements" to "items" (to be less list-y)
+}
+
+func newIter[T KredisTyped](elements []T) iterator[T] {
+	return &iter[T]{elements}
+}
+
+func (i *iter[T]) next() (elem T, ok bool) {
+	if len(i.elements) == 0 {
+		return // no elements - return empty values
+	}
+
+	elem = i.elements[0]
+	i.elements = i.elements[1:]
+
+	return elem, true
+}
+
+func (i *iter[T]) values() []interface{} {
+	values := make([]interface{}, len(i.elements))
+
+	for idx, e := range i.elements {
+		values[idx] = typeToInterface(e)
+	}
+
+	return values
+}
+
+func (i *iter[T]) unique() []interface{} {
+	m := make(map[T]struct{})
+	values := make([]interface{}, 0)
+
+	for _, e := range i.elements {
+		if _, exists := m[e]; exists {
+			continue
+		}
+
+		m[e] = struct{}{}
+		values = append(values, typeToInterface(e))
+	}
+
+	return values
+}
+
+func (i *iter[T]) uniqueMap() (m map[T]struct{}) {
+	if len(i.elements) == 0 {
+		return
+	}
+
+	m = make(map[T]struct{})
+	for _, e := range i.elements {
+		m[e] = struct{}{}
+	}
+
+	return
+}

iter_test.go

@@ -0,0 +1,27 @@
+package kredis
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestIterValues(t *testing.T) {
+	i := newIter([]kredisJSON{
+		*NewKredisJSON(`{"a":"b"}`),
+		*NewKredisJSON(`{"c":"d"}`),
+		*NewKredisJSON(`{"a":"b"}`),
+	})
+
+	assert.Len(t, i.unique(), 2)
+}
+
+func TestIterUnique(t *testing.T) {
+	i := newIter([]kredisJSON{
+		*NewKredisJSON(`{"a":"b"}`),
+		*NewKredisJSON(`{"c":"d"}`),
+		*NewKredisJSON(`{"a":"b"}`),
+	})
+
+	assert.Len(t, i.unique(), 2)
+}

kredis.go

@@ -2,32 +2,123 @@ package kredis
 
 import (
 	"encoding/json"
+	"strconv"
 	"time"
+
+	"github.com/redis/go-redis/v9"
 )
 
 // TODO does this need to be exported??
-type kredisJSON []byte
+// type kredisJSON []byte
+type kredisJSON struct {
+	s string
+}
 
 type KredisTyped interface {
+	comparable
 	~bool | ~int | ~string | kredisJSON | time.Time
 }
 
 func NewKredisJSON(jsonStr string) *kredisJSON {
-	var kj kredisJSON = kredisJSON(jsonStr)
+	var kj kredisJSON = kredisJSON{jsonStr}
 
 	return &kj
 }
 
 func (kj kredisJSON) String() string {
-	return string(kj)
+	return kj.s
 }
 
 func (kj *kredisJSON) Unmarshal(data *interface{}) error {
-	err := json.Unmarshal(*kj, data)
-
+	err := json.Unmarshal([]byte(kj.s), data)
 	if err != nil {
 		return err
 	}
 
 	return nil
 }
+
+func typeToInterface[T KredisTyped](t T) interface{} {
+	switch any(t).(type) {
+	case time.Time:
+		return any(t).(time.Time).Format(time.RFC3339Nano)
+	case kredisJSON:
+		return any(t).(kredisJSON).String()
+	default:
+		return t
+	}
+}
+
+func stringCmdToTyped[T KredisTyped](cmd *redis.StringCmd, typed *T) (T, bool) {
+	if cmd.Err() == redis.Nil {
+		// return any(*typed).(T), false
+		goto Empty
+	}
+
+	switch any(*typed).(type) {
+	case bool:
+		b, err := cmd.Bool()
+		if err != nil {
+			return any(false).(T), false
+		}
+		return any(b).(T), true
+
+	case int:
+		n, err := cmd.Int()
+		if err != nil {
+			return any(0).(T), false
+		}
+		return any(n).(T), true
+
+	case time.Time:
+		t, err := cmd.Time()
+		if err != nil {
+			return any(time.Time{}).(T), false
+		}
+		return any(t).(T), true
+
+	case string:
+		return any(cmd.Val()).(T), true
+
+	case kredisJSON:
+		return any(NewKredisJSON(cmd.Val())).(T), true
+	}
+
+Empty:
+	return any(*typed).(T), false
+}
+
+func copyCmdSliceTo[T KredisTyped](slice []interface{}, dst []T) int {
+	var total int
+
+	for i, e := range slice {
+		if i == len(dst) {
+			break
+		}
+
+		switch any(dst[i]).(type) {
+		case bool:
+			b, _ := strconv.ParseBool(e.(string))
+
+			dst[i] = any(b).(T)
+		case int:
+			n, _ := strconv.Atoi(e.(string))
+
+			dst[i] = any(n).(T)
+		case kredisJSON:
+			j := NewKredisJSON(e.(string))
+
+			dst[i] = any(*j).(T)
+		case time.Time:
+			t, _ := time.Parse(time.RFC3339Nano, e.(string))
+
+			dst[i] = any(t).(T)
+		default:
+			dst[i] = any(e).(T)
+		}
+
+		total = total + 1
+	}
+
+	return total
+}