From a2f805f05177ccf7b1eb749a858483abfd2d9e37 Mon Sep 17 00:00:00 2001 From: ayushsatyam146 Date: Tue, 13 May 2025 12:42:04 +0530 Subject: [PATCH] Initializing thick client --- examples/example_ping/main.go | 3 +- examples/example_watch/main.go | 61 ++++++++++++++++++++++ go.mod | 9 ++++ go.sum | 12 +++++ main.go | 94 +++++++++++++++++++++++++++++++++- main_test.go | 5 +- 6 files changed, 180 insertions(+), 4 deletions(-) create mode 100644 examples/example_watch/main.go diff --git a/examples/example_ping/main.go b/examples/example_ping/main.go index c633988..226f2b6 100644 --- a/examples/example_ping/main.go +++ b/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) } diff --git a/examples/example_watch/main.go b/examples/example_watch/main.go new file mode 100644 index 0000000..bea2b76 --- /dev/null +++ b/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() +} diff --git a/go.mod b/go.mod index d66e3e4..5a30fdb 100644 --- a/go.mod +++ b/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 +) diff --git a/go.sum b/go.sum index cc36f44..b2ff53a 100644 --- a/go.sum +++ b/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= diff --git a/main.go b/main.go index 186c92b..9f621be 100644 --- a/main.go +++ b/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)) +} diff --git a/main_test.go b/main_test.go index fb8c751..164d50e 100644 --- a/main_test.go +++ b/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) }