initializing thick client

Author: ayushsatyam146

examples/example_ping/main.go

@@ -15,4 +15,15 @@ func main() {
 
 	resp := client.Fire(&wire.Command{Cmd: "PING"})
 	fmt.Println(resp)
-}
+
+	resp = client.Fire(&wire.Command{Cmd: "SET", Args: []string{"foo", "bar"}})
+	fmt.Println(resp)
+
+	resp = client.Fire(&wire.Command{Cmd: "GET", Args: []string{"foo"}})
+	fmt.Println(resp)
+
+	resp = client.Fire(&wire.Command{Cmd: "GET", Args: []string{"foo1"}})
+	fmt.Println(resp)
+
+	client.Close()
+}

go.mod

@@ -4,4 +4,15 @@ go 1.24.0
 
 require google.golang.org/protobuf v1.36.5
 
-require github.com/google/uuid v1.6.0
+require (
+	github.com/dgraph-io/ristretto v0.2.0
+	github.com/dgryski/go-farm v0.0.0-20240924180020-3414d57e47da
+	github.com/google/uuid v1.6.0
+)
+
+require (
+	github.com/cespare/xxhash/v2 v2.1.1 // 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,8 +1,28 @@
+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/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
+github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+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=
+github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
+github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/stretchr/testify v1.8.4 h1:CcVxjf3Q8PM0mHUKJCdn+eZZtm5yQwehR5yeSVQQcUk=
+github.com/stretchr/testify v1.8.4/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo=
+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.4 h1:6A3ZDJHn/eNqc1i+IdefRzy/9PokBTPvcqMySR7NNIM=
-google.golang.org/protobuf v1.36.4/go.mod h1:9fA7Ob0pmnwhb644+1+CVWFRbNajQ6iRojtC/QF5bRE=
+google.golang.org/protobuf v1.36.5 h1:tPhr+woSbjfYvY6/GPufUoYizxw1cF/yFoxJ2fmpwlM=
+google.golang.org/protobuf v1.36.5/go.mod h1:9fA7Ob0pmnwhb644+1+CVWFRbNajQ6iRojtC/QF5bRE=
+gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
+gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

main.go

@@ -3,12 +3,17 @@ package dicedb
 import (
 	"fmt"
 	"net"
+	"strconv"
 	"strings"
+	"sync"
 	"time"
 
+	"github.com/dgraph-io/ristretto"
+	"github.com/dgryski/go-farm"
 	"github.com/dicedb/dicedb-go/ironhawk"
 	"github.com/dicedb/dicedb-go/wire"
 	"github.com/google/uuid"
+	"google.golang.org/protobuf/types/known/structpb"
 )
 
 type Client struct {
@@ -18,6 +23,9 @@ type Client struct {
 	watchCh   chan *wire.Response
 	host      string
 	port      int
+	lcache    *ristretto.Cache
+	mu        sync.RWMutex // Mutex for thread-safe operations
+	wg        sync.WaitGroup
 }
 
 type option func(*Client)
@@ -43,7 +51,17 @@ func NewClient(host string, port int, opts ...option) (*Client, error) {
 		return nil, err
 	}
 
-	client := &Client{conn: conn, host: host, port: port}
+	// Initialize Ristretto cache
+	cache, err := ristretto.NewCache(&ristretto.Config{
+		NumCounters: 1e7,     // number of keys to track frequency of (10M)
+		MaxCost:     1 << 30, // maximum cost of cache (1GB)
+		BufferItems: 64,      // number of keys per Get buffer
+	})
+	if err != nil {
+		return nil, fmt.Errorf("failed to create Ristretto cache: %w", err)
+	}
+
+	client := &Client{conn: conn, host: host, port: port, lcache: cache}
 	for _, opt := range opts {
 		opt(client)
 	}
@@ -59,6 +77,14 @@ func NewClient(host string, port int, opts ...option) (*Client, error) {
 		return nil, fmt.Errorf("could not complete the handshake: %s", resp.Err)
 	}
 
+	client.wg.Add(1)
+	go func(client *Client) {
+		fmt.Println("Listening for messages")
+		ListenForMessages(client, func(message string) {
+		})
+		client.wg.Done()
+	}(client)
+
 	return client, nil
 }
 
@@ -80,6 +106,44 @@ func (c *Client) fire(cmd *wire.Command, co net.Conn) *wire.Response {
 }
 
 func (c *Client) Fire(cmd *wire.Command) *wire.Response {
+	// TODO: handle this better by refining read-only commands
+	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 the key is in the cache
+		if value, found := c.lcache.Get(fp); found {
+			return &wire.Response{
+				Value: &wire.Response_VStr{VStr: value.(string)},
+			}
+		}
+
+		// If not in cache, send the command to the server
+		resp := c.fire(cmd, c.conn)
+
+		// If the response is successful, store it in the cache
+		// and subscribe to the key to keep the value updated
+		if resp.Err == "" {
+			c.lcache.Set(fp, resp.GetVStr(), 1) // Store only the string value
+			c.lcache.Wait()                     // Ensure value is stored before proceeding
+
+			c.wg.Add(1)
+			go func() {
+				Subscribe(c, key)
+				c.wg.Done()
+			}()
+		}
+
+		return resp
+	}
+
+	// For non-GET commands, just send to the server
 	return c.fire(cmd, c.conn)
 }
 
@@ -137,5 +201,51 @@ func (c *Client) watch() {
 }
 
 func (c *Client) Close() {
+	c.wg.Wait()
 	c.conn.Close()
 }
+
+// SubscriptionManager methods
+func Subscribe(client *Client, watch_key string) {
+	resp := client.Fire(&wire.Command{
+		Cmd:  "GET.WATCH",
+		Args: []string{watch_key},
+	})
+	if resp.Err != "" {
+		fmt.Println("error subscribing:", resp.Err)
+	}
+}
+
+func ListenForMessages(client *Client, onMessage func(message string)) {
+	ch, err := client.WatchCh()
+	if err != nil {
+		panic(err)
+	}
+	for resp := range ch {
+		var fp uint32
+		// Extract fingerprint from string_value if available
+		if resp.Attrs != nil {
+			if fpValue, ok := resp.Attrs.Fields["fingerprint"]; ok {
+				if fpString, ok := fpValue.Kind.(*structpb.Value_StringValue); ok {
+					// Convert string fingerprint to uint32
+					fpUint, err := strconv.ParseUint(fpString.StringValue, 10, 32)
+					if err == nil {
+						fp = uint32(fpUint)
+					}
+				}
+			}
+		}
+
+		client.lcache.Set(fp, resp.GetVStr(), 1) // Store only the string value
+		client.lcache.Wait()                     // Ensure value is stored before proceeding
+		onMessage(resp.GetVStr())
+	}
+}
+
+func Fingerprint(c *wire.Command) uint32 {
+	cmdStr := c.Cmd
+	if len(c.Args) > 0 {
+		cmdStr = fmt.Sprintf("%s %s", c.Cmd, strings.Join(c.Args, " "))
+	}
+	return farm.Fingerprint32([]byte(cmdStr))
+}