Initializing thick client

examples/example_ping/main.go

@@ -2,13 +2,14 @@ package main
 
 import (
 	"fmt"
+	"sync"
 
 	"github.com/dicedb/dicedb-go"
 	"github.com/dicedb/dicedb-go/wire"
 )
 
 func main() {
-	client, err := dicedb.NewClient("localhost", 7379)
+	client, err := dicedb.NewClient("localhost", 7379, &sync.WaitGroup{})
 	if err != nil {
 		fmt.Println(err)
 	}

examples/example_watch/main.go

@@ -0,0 +1,61 @@
+package main
+
+import (
+	"fmt"
+	"sync"
+	"time"
+
+	"github.com/dicedb/dicedb-go"
+	"github.com/dicedb/dicedb-go/wire"
+)
+
+func main() {
+	var wg sync.WaitGroup
+
+	client, err := dicedb.NewClient("localhost", 7379, &wg)
+	if err != nil {
+		fmt.Println(err)
+	}
+
+	resp := client.Fire(&wire.Command{Cmd: "PING"})
+	fmt.Println(resp)
+
+	wg.Add(1)
+	go func(client *dicedb.Client) {
+		dicedb.ListenForMessages(client, func(message string) {
+			fmt.Println("Received message from second watcher:", message)
+		})
+		wg.Done()
+	}(client)
+
+	resp = client.Fire(&wire.Command{Cmd: "SET", Args: []string{"k1", "v1"}})
+	fmt.Println(resp)
+
+	resp = client.Fire(&wire.Command{Cmd: "GET", Args: []string{"k1"}})
+	fmt.Println(resp)
+
+	resp = client.Fire(&wire.Command{Cmd: "SET", Args: []string{"k2", "v2"}})
+	fmt.Println(resp)
+
+	resp = client.Fire(&wire.Command{Cmd: "GET", Args: []string{"k2"}})
+	fmt.Println(resp)
+
+	resp = client.Fire(&wire.Command{Cmd: "SET", Args: []string{"k3", "v3"}})
+	fmt.Println(resp)
+
+	resp = client.Fire(&wire.Command{Cmd: "GET", Args: []string{"k3"}})
+	fmt.Println(resp)
+
+	time.Sleep(30 * time.Second)
+
+	resp1 := client.Fire(&wire.Command{Cmd: "GET", Args: []string{"k1"}})
+	fmt.Println("waited k1 ", resp1)
+
+	resp2 := client.Fire(&wire.Command{Cmd: "GET", Args: []string{"k2"}})
+	fmt.Println("waited k2 ", resp2)
+
+	resp3 := client.Fire(&wire.Command{Cmd: "GET", Args: []string{"k3"}})
+	fmt.Println("waited k3 ", resp3)
+
+	wg.Wait()
+}

go.mod

@@ -8,3 +8,12 @@ require (
 	github.com/google/uuid v1.6.0
 	go.uber.org/mock v0.5.1
 )
+
+require (
+	github.com/cespare/xxhash/v2 v2.1.1 // indirect
+	github.com/dgraph-io/ristretto v0.2.0 // indirect
+	github.com/dgryski/go-farm v0.0.0-20240924180020-3414d57e47da // indirect
+	github.com/dustin/go-humanize v1.0.1 // indirect
+	github.com/pkg/errors v0.9.1 // indirect
+	golang.org/x/sys v0.11.0 // indirect
+)

go.sum

@@ -1,9 +1,21 @@
+github.com/cespare/xxhash/v2 v2.1.1 h1:6MnRN8NT7+YBpUIWxHtefFZOKTAPgGjpQSxqLNn0+qY=
+github.com/cespare/xxhash/v2 v2.1.1/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
+github.com/dgraph-io/ristretto v0.2.0 h1:XAfl+7cmoUDWW/2Lx8TGZQjjxIQ2Ley9DSf52dru4WE=
+github.com/dgraph-io/ristretto v0.2.0/go.mod h1:8uBHCU/PBV4Ag0CJrP47b9Ofby5dqWNh4FicAdoqFNU=
+github.com/dgryski/go-farm v0.0.0-20240924180020-3414d57e47da h1:aIftn67I1fkbMa512G+w+Pxci9hJPB8oMnkcP3iZF38=
+github.com/dgryski/go-farm v0.0.0-20240924180020-3414d57e47da/go.mod h1:SqUrOPUnsFjfmXRMNPybcSiG0BgUW2AuFH8PAnS2iTw=
+github.com/dustin/go-humanize v1.0.1 h1:GzkhY7T5VNhEkwH0PVJgjz+fX1rhBrR7pRT3mDkpeCY=
+github.com/dustin/go-humanize v1.0.1/go.mod h1:Mu1zIs6XwVuF/gI1OepvI0qD18qycQx+mFykh5fBlto=
 github.com/google/go-cmp v0.5.5 h1:Khx7svrCpmxxtHBq5j2mp/xVjsi8hQMfNLvJFAlrGgU=
 github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
 github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
 github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
+github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
+github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 go.uber.org/mock v0.5.1 h1:ASgazW/qBmR+A32MYFDB6E2POoTgOwT509VP0CT/fjs=
 go.uber.org/mock v0.5.1/go.mod h1:ge71pBPLYDk7QIi1LupWxdAykm7KIEFchiOqd6z7qMM=
+golang.org/x/sys v0.11.0 h1:eG7RXZHdqOJ1i+0lgLgCpSXAp6M3LYlAo6osgSi0xOM=
+golang.org/x/sys v0.11.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543 h1:E7g+9GITq07hpfrRu66IVDexMakfv52eLZ2CXBWiKr4=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 google.golang.org/protobuf v1.36.5 h1:tPhr+woSbjfYvY6/GPufUoYizxw1cF/yFoxJ2fmpwlM=

main.go

@@ -7,6 +7,8 @@ import (
 	"sync"
 	"time"
 
+	"github.com/dgraph-io/ristretto"
+	"github.com/dgryski/go-farm"
 	"github.com/dicedb/dicedb-go/wire"
 	"github.com/google/uuid"
 )
@@ -22,7 +24,9 @@ type Client struct {
 	watchWire    *ClientWire
 	watchCh      chan *wire.Result
 	host         string
+	lcache       *ristretto.Cache
 	port         int
+	wg           *sync.WaitGroup
 }
 
 type option func(*Client)
@@ -33,7 +37,7 @@ func WithID(id string) option {
 	}
 }
 
-func NewClient(host string, port int, opts ...option) (*Client, error) {
+func NewClient(host string, port int, wg *sync.WaitGroup, opts ...option) (*Client, error) {
 	mainRetrier := NewRetrier(3, 5*time.Second)
 	clientWire, err := ExecuteWithResult(mainRetrier, []wire.ErrKind{wire.NotEstablished}, func() (*ClientWire, *wire.WireError) {
 		return NewClientWire(maxResponseSize, host, port)
@@ -47,11 +51,22 @@ func NewClient(host string, port int, opts ...option) (*Client, error) {
 		return nil, fmt.Errorf("unexpected error when establishing server connection, report this to dicedb maintainers: %w", err)
 	}
 
+	lru_cache, cacheErr := ristretto.NewCache(&ristretto.Config{
+		NumCounters: 1e6,     // number of keys to track frequency of (1M).
+		MaxCost:     1 << 30, // maximum cost of cache (1GB).
+		BufferItems: 64,      // number of keys per Get buffer.
+	})
+	if cacheErr != nil {
+		return nil, fmt.Errorf("failed to create local cache: %w", cacheErr)
+	}
+
 	client := &Client{
 		mainRetrier: mainRetrier,
 		mainWire:    clientWire,
 		host:        host,
 		port:        port,
+		wg:          wg,
+		lcache:      lru_cache,
 	}
 
 	for _, opt := range opts {
@@ -110,6 +125,42 @@ func (c *Client) fire(cmd *wire.Command, clientWire *ClientWire) *wire.Result {
 }
 
 func (c *Client) Fire(cmd *wire.Command) *wire.Result {
+	if cmd.Cmd == "GET" && len(cmd.Args) > 0 {
+		key := cmd.Args[0]
+
+		watch_cmd := &wire.Command{
+			Cmd:  "GET.WATCH",
+			Args: []string{key},
+		}
+
+		fp := Fingerprint(watch_cmd)
+
+		// check if fingerprint key is in cache
+		if cachedValue, found := c.lcache.Get(fp); found {
+			if cachedResult, ok := cachedValue.(*wire.Result); ok {
+				return cachedResult
+			}
+		}
+
+		// if it is not, fire the the normal get command
+		resp := c.fire(cmd, c.mainWire)
+
+		// if the get command was successful, store the value in cache
+		// and subscribe to the key
+		if resp.Status == wire.Status_OK {
+			c.lcache.Set(fp, resp, 1)
+			c.lcache.Wait()
+
+			c.wg.Add(1)
+			go func() {
+				defer c.wg.Done()
+				Subscribe(c, cmd.Args[0])
+			}()
+		}
+		return resp
+	}
+
+	// For non-GET commands, just send to the server
 	return c.fire(cmd, c.mainWire)
 }
 
@@ -202,3 +253,44 @@ func (c *Client) restoreWire(dst *ClientWire) *wire.WireError { // nolint:static
 func noop() *wire.WireError {
 	return nil
 }
+
+func Subscribe(client *Client, watch_key string) {
+	resp := client.Fire(&wire.Command{
+		Cmd:  "GET.WATCH",
+		Args: []string{watch_key},
+	})
+	if resp.Status == wire.Status_ERR {
+		fmt.Errorf("error subscribing:", resp.Message)
+	}
+}
+
+func ListenForMessages(client *Client, onMessage func(message string)) {
+	ch, err := client.WatchCh()
+	if err != nil {
+		panic(err)
+	}
+	for resp := range ch {
+		// take fingerprint of the resp and check if it is in cache
+		// if it is in cache then update the Response part of the cache entry
+		fp := resp.GetFingerprint64()
+		if cachedValue, found := client.lcache.Get(fp); found {
+			if cachedResult, ok := cachedValue.(*wire.Result); ok {
+				cachedResult.Response = resp.Response
+				client.lcache.Set(fp, cachedResult, 1)
+			}
+		}
+		if resp.Status == wire.Status_ERR {
+			fmt.Errorf("error listening for messages:", resp.Message)
+		} else {
+			onMessage(resp.GetMessage())
+		}
+	}
+}
+
+func Fingerprint(c *wire.Command) uint64 {
+	cmdStr := c.Cmd
+	if len(c.Args) > 0 {
+		cmdStr = fmt.Sprintf("%s %s", c.Cmd, strings.Join(c.Args, " "))
+	}
+	return farm.Fingerprint64([]byte(cmdStr))
+}

main_test.go

@@ -2,6 +2,7 @@ package dicedb
 
 import (
 	"errors"
+	"sync"
 	"testing"
 
 	"github.com/dicedb/dicedb-go/wire"
@@ -40,7 +41,7 @@ func TestNewClient(t *testing.T) {
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			client, err := NewClient(tt.host, tt.port)
+			client, err := NewClient(tt.host, tt.port, &sync.WaitGroup{})
 			if (client == nil) != tt.wantNil {
 				t.Errorf("NewClient() got = %v, %s, want nil = %v, err = %v", client, err, tt.wantNil, tt.err)
 			}
@@ -52,7 +53,7 @@ func TestNewClient(t *testing.T) {
 }
 
 func TestClient_Fire(t *testing.T) {
-	client, err := NewClient("localhost", 7379)
+	client, err := NewClient("localhost", 7379, &sync.WaitGroup{})
 	if err != nil {
 		t.Fatalf("failed to create client: %v", err)
 	}