A examples directory with recipes?

[KennyHoang-CS]

Thoughts of an example directory with recipes like:

1. set up pub/sub
2. set up valkey streams
3. etc

So far, it looks like I need to look at the func doc comments if I were to put something together using valkey-go.

I'm interested to see if I can set up valkey-go as an external data store via aws elasticcache (w/ the valkey engine).

The problem I run to at the moment before valkey is that .. I have an app that uses golang channels to handle messages in the app's internal memory, so if it the app goes down (I lose state of the messages), would valkey-go solve this? Hence, I was looking to see if there is a valkey-go example/recipe that can handle this.

[rueian]

Hi @KennyHoang-CS, sounds like you need a queue with some persistence. Usually, Valkey Pub/Sub can't be used as a queue because a consumer will never get messages that were published during its disconnection.

Valkey Stream can be used as a queue, but it is super complicated. I will just suggest you use Valkey List (LPUSH+BRPOP).

If you can provide how your current application uses the go channels, we may come up with better recipes.

[KennyHoang-CS]

So, we have a backend server that basically does this:

1. request comes in with an id, performs some logic A, send that id into step 2 via a go routine, and returns a response back on the outcome of logic A (nil or error). Step 2 happens async.
2. go routine publishes the request id into a go channel
3. perform some transaction database operations using that id
4. if any db operations fail, the id gets sent back into the go channel to repeat step 3 until it succeeds
5. done

We have high volume of requests as well, and messages would be lost in the go channel if the app goes down or disconnects. So, we would need some form of persistence. Similar to aws sqs, or kafka.

[KennyHoang-CS]

@rueian thoughts?

[rueian]

Hi @KennyHoang-CS, this is a common example:

package main

import (
	"context"
	"fmt"
	"log"
	"strconv"
	"time"

	valkey "github.com/valkey-io/valkey-go"
)

const queueKey = "work:queue"

func Enqueue(ctx context.Context, c valkey.Client, id string) error {
	return c.Do(ctx, c.B().Lpush().Key(queueKey).Element(id).Build()).Error()
}

func worker(ctx context.Context, c valkey.Client, wid int) {
	for {
		// Block forever for the next item
		resp := c.Do(ctx, c.B().Brpop().Key(queueKey).Timeout(0).Build())
		parts, err := resp.AsStrSlice()
		if err != nil || len(parts) != 2 {
			log.Printf("[worker %d] unexpected reply: %#v err=%v", wid, parts, err)
			continue
		}
		id := parts[1]
		fmt.Printf("[worker %d] processing %s\n", wid, id)

		// do your DB work here...
	}
}

func main() {
	// Connect to Valkey
	client, err := valkey.NewClient(valkey.ClientOption{InitAddress: []string{"127.0.0.1:6379"}})
	if err != nil {
		log.Fatal(err)
	}
	defer client.Close()

	ctx := context.Background()

	// Start N workers
	const workerCount = 4
	for i := 0; i < workerCount; i++ {
		go worker(ctx, client, i)
	}

	// Publisher: enqueue IDs forever
	for i := 1; true; i++ {
		if err := Enqueue(ctx, client, strconv.Itoa(i)); err != nil {
			log.Printf("enqueue error: %v", err)
		}
		time.Sleep(500 * time.Millisecond) // adjust as needed
	}
}