feat: Port to new Dudirekta version

pojntfx · Nov 9, 2023 · b7c39f7 · b7c39f7
1 parent 421d1e1
commit b7c39f7
Show file tree

Hide file tree

Showing 7 changed files with 147 additions and 36 deletions.
diff --git a/cmd/r3map-benchmark-direct-mount/main.go b/cmd/r3map-benchmark-direct-mount/main.go
@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"context"
 	"crypto/rand"
+	"encoding/json"
 	"errors"
 	"flag"
 	"fmt"
@@ -138,14 +139,12 @@ func main() {
 
 		case backendTypeDudirekta:
 			ready := make(chan struct{})
-			registry := rpc.NewRegistry(
-				&struct{}{},
-				services.BackendRemote{},
+			registry := rpc.NewRegistry[services.BackendRemote, json.RawMessage](
+				struct{}{},
 
 				time.Second*10,
 				ctx,
 				&rpc.Options{
-					ResponseBufferLen: rpc.DefaultResponseBufferLen,
 					OnClientConnect: func(remoteID string) {
 						ready <- struct{}{}
 					},
@@ -159,7 +158,29 @@ func main() {
 			defer conn.Close()
 
 			go func() {
-				if err := registry.Link(conn); err != nil {
+				encoder := json.NewEncoder(conn)
+				decoder := json.NewDecoder(conn)
+
+				if err := registry.LinkStream(
+					func(v rpc.Message[json.RawMessage]) error {
+						return encoder.Encode(v)
+					},
+					func(v *rpc.Message[json.RawMessage]) error {
+						return decoder.Decode(v)
+					},
+
+					func(v any) (json.RawMessage, error) {
+						b, err := json.Marshal(v)
+						if err != nil {
+							return nil, err
+						}
+
+						return json.RawMessage(b), nil
+					},
+					func(data json.RawMessage, v any) error {
+						return json.Unmarshal([]byte(data), v)
+					},
+				); err != nil {
 					if !utils.IsClosedErr(err) {
 						panic(err)
 					}
@@ -169,11 +190,12 @@ func main() {
 			<-ready
 
 			var peer *services.BackendRemote
-			for _, candidate := range registry.Peers() {
-				peer = &candidate
 
-				break
-			}
+			_ = registry.ForRemotes(func(remoteID string, remote services.BackendRemote) error {
+				peer = &remote
+
+				return nil
+			})
 
 			if peer == nil {
 				panic(errNoPeerFound)

diff --git a/cmd/r3map-benchmark-managed-mount/main.go b/cmd/r3map-benchmark-managed-mount/main.go
@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"context"
 	"crypto/rand"
+	"encoding/json"
 	"errors"
 	"flag"
 	"fmt"
@@ -162,14 +163,12 @@ func main() {
 
 		case backendTypeDudirekta:
 			ready := make(chan struct{})
-			registry := rpc.NewRegistry(
-				&struct{}{},
-				services.BackendRemote{},
+			registry := rpc.NewRegistry[services.BackendRemote, json.RawMessage](
+				struct{}{},
 
 				time.Second*10,
 				ctx,
 				&rpc.Options{
-					ResponseBufferLen: rpc.DefaultResponseBufferLen,
 					OnClientConnect: func(remoteID string) {
 						ready <- struct{}{}
 					},
@@ -183,7 +182,29 @@ func main() {
 			defer conn.Close()
 
 			go func() {
-				if err := registry.Link(conn); err != nil {
+				encoder := json.NewEncoder(conn)
+				decoder := json.NewDecoder(conn)
+
+				if err := registry.LinkStream(
+					func(v rpc.Message[json.RawMessage]) error {
+						return encoder.Encode(v)
+					},
+					func(v *rpc.Message[json.RawMessage]) error {
+						return decoder.Decode(v)
+					},
+
+					func(v any) (json.RawMessage, error) {
+						b, err := json.Marshal(v)
+						if err != nil {
+							return nil, err
+						}
+
+						return json.RawMessage(b), nil
+					},
+					func(data json.RawMessage, v any) error {
+						return json.Unmarshal([]byte(data), v)
+					},
+				); err != nil {
 					if !utils.IsClosedErr(err) {
 						panic(err)
 					}
@@ -193,11 +214,12 @@ func main() {
 			<-ready
 
 			var peer *services.BackendRemote
-			for _, candidate := range registry.Peers() {
-				peer = &candidate
 
-				break
-			}
+			_ = registry.ForRemotes(func(remoteID string, remote services.BackendRemote) error {
+				peer = &remote
+
+				return nil
+			})
 
 			if peer == nil {
 				panic(errNoPeerFound)

diff --git a/cmd/r3map-benchmark-migration-server/main.go b/cmd/r3map-benchmark-migration-server/main.go
@@ -4,6 +4,7 @@ import (
 	"bufio"
 	"context"
 	"crypto/rand"
+	"encoding/json"
 	"errors"
 	"flag"
 	"fmt"
@@ -258,14 +259,12 @@ func main() {
 		}()
 	} else {
 		clients := 0
-		registry := rpc.NewRegistry(
+		registry := rpc.NewRegistry[struct{}, json.RawMessage](
 			svc,
-			struct{}{},
 
 			time.Second*10,
 			ctx,
 			&rpc.Options{
-				ResponseBufferLen: rpc.DefaultResponseBufferLen,
 				OnClientConnect: func(remoteID string) {
 					clients++
 
@@ -309,7 +308,29 @@ func main() {
 						}
 					}()
 
-					if err := registry.Link(conn); err != nil {
+					encoder := json.NewEncoder(conn)
+					decoder := json.NewDecoder(conn)
+
+					if err := registry.LinkStream(
+						func(v rpc.Message[json.RawMessage]) error {
+							return encoder.Encode(v)
+						},
+						func(v *rpc.Message[json.RawMessage]) error {
+							return decoder.Decode(v)
+						},
+
+						func(v any) (json.RawMessage, error) {
+							b, err := json.Marshal(v)
+							if err != nil {
+								return nil, err
+							}
+
+							return json.RawMessage(b), nil
+						},
+						func(data json.RawMessage, v any) error {
+							return json.Unmarshal([]byte(data), v)
+						},
+					); err != nil {
 						panic(err)
 					}
 				}()

diff --git a/cmd/r3map-benchmark-migration/main.go b/cmd/r3map-benchmark-migration/main.go
@@ -5,6 +5,7 @@ import (
 	"bytes"
 	"context"
 	"crypto/rand"
+	"encoding/json"
 	"errors"
 	"flag"
 	"fmt"
@@ -216,22 +217,42 @@ func main() {
 		defer conn.Close()
 
 		ready := make(chan struct{})
-		registry := rpc.NewRegistry(
+		registry := rpc.NewRegistry[services.SeederRemote, json.RawMessage](
 			&struct{}{},
-			services.SeederRemote{},
 
 			time.Second*10,
 			ctx,
 			&rpc.Options{
-				ResponseBufferLen: rpc.DefaultResponseBufferLen,
 				OnClientConnect: func(remoteID string) {
 					ready <- struct{}{}
 				},
 			},
 		)
 
 		go func() {
-			if err := registry.Link(conn); err != nil {
+			encoder := json.NewEncoder(conn)
+			decoder := json.NewDecoder(conn)
+
+			if err := registry.LinkStream(
+				func(v rpc.Message[json.RawMessage]) error {
+					return encoder.Encode(v)
+				},
+				func(v *rpc.Message[json.RawMessage]) error {
+					return decoder.Decode(v)
+				},
+
+				func(v any) (json.RawMessage, error) {
+					b, err := json.Marshal(v)
+					if err != nil {
+						return nil, err
+					}
+
+					return json.RawMessage(b), nil
+				},
+				func(data json.RawMessage, v any) error {
+					return json.Unmarshal([]byte(data), v)
+				},
+			); err != nil {
 				if !utils.IsClosedErr(err) {
 					seederErrs <- err
 
@@ -244,11 +265,11 @@ func main() {
 
 		<-ready
 
-		for _, candidate := range registry.Peers() {
-			peer = &candidate
+		_ = registry.ForRemotes(func(remoteID string, remote services.SeederRemote) error {
+			peer = &remote
 
-			break
-		}
+			return nil
+		})
 
 		if peer == nil {
 			panic(errNoPeerFound)

diff --git a/cmd/r3map-benchmark-mount-server/main.go b/cmd/r3map-benchmark-mount-server/main.go
@@ -2,6 +2,7 @@ package main
 
 import (
 	"context"
+	"encoding/json"
 	"errors"
 	"flag"
 	"fmt"
@@ -145,14 +146,12 @@ func main() {
 		}()
 	} else {
 		clients := 0
-		registry := rpc.NewRegistry(
+		registry := rpc.NewRegistry[struct{}, json.RawMessage](
 			svc,
-			struct{}{},
 
 			time.Second*10,
 			ctx,
 			&rpc.Options{
-				ResponseBufferLen: rpc.DefaultResponseBufferLen,
 				OnClientConnect: func(remoteID string) {
 					clients++
 
@@ -196,7 +195,29 @@ func main() {
 						}
 					}()
 
-					if err := registry.Link(conn); err != nil {
+					encoder := json.NewEncoder(conn)
+					decoder := json.NewDecoder(conn)
+
+					if err := registry.LinkStream(
+						func(v rpc.Message[json.RawMessage]) error {
+							return encoder.Encode(v)
+						},
+						func(v *rpc.Message[json.RawMessage]) error {
+							return decoder.Decode(v)
+						},
+
+						func(v any) (json.RawMessage, error) {
+							b, err := json.Marshal(v)
+							if err != nil {
+								return nil, err
+							}
+
+							return json.RawMessage(b), nil
+						},
+						func(data json.RawMessage, v any) error {
+							return json.Unmarshal([]byte(data), v)
+						},
+					); err != nil {
 						panic(err)
 					}
 				}()

diff --git a/go.mod b/go.mod
@@ -9,9 +9,9 @@ require (
 	github.com/loopholelabs/frisbee-go v0.7.1
 	github.com/loopholelabs/polyglot-go v0.5.1
 	github.com/minio/minio-go v6.0.14+incompatible
-	github.com/pojntfx/dudirekta v0.5.1
+	github.com/pojntfx/dudirekta v0.6.0
 	github.com/pojntfx/go-nbd v0.3.2
-	github.com/redis/go-redis/v9 v9.0.5
+	github.com/redis/go-redis/v9 v9.2.1
 	github.com/rs/zerolog v1.30.0
 	github.com/schollz/progressbar/v3 v3.13.1
 	google.golang.org/grpc v1.57.0

diff --git a/go.sum b/go.sum
@@ -72,10 +72,14 @@ github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZb
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/pojntfx/dudirekta v0.5.1 h1:omrLk+lFJTQd6F/FGhW/tXEJF6Q/Zb4x++6zgWu8Ahw=
 github.com/pojntfx/dudirekta v0.5.1/go.mod h1:2G79XDOe1c3Nz3G+LQfiNZ5K/SS3b2TP1K9JyRt8woI=
+github.com/pojntfx/dudirekta v0.6.0 h1:bNO5jrHDCSiVf+YCjU6GiEyY7S9/KtgzsGC0KruobUc=
+github.com/pojntfx/dudirekta v0.6.0/go.mod h1:zNquHCePC+378rWnJ0SWs1p9qdZdrJ4PFGNyVdQWD1U=
 github.com/pojntfx/go-nbd v0.3.2 h1:qI6S4qsHD87V9fTH6jiS4DIqq/rWmI0El0xSToMUDeg=
 github.com/pojntfx/go-nbd v0.3.2/go.mod h1:SehHnbi2e8NiSAKby42Itm8SIoS7b+wAprsfPH3qgYk=
 github.com/redis/go-redis/v9 v9.0.5 h1:CuQcn5HIEeK7BgElubPP8CGtE0KakrnbBSTLjathl5o=
 github.com/redis/go-redis/v9 v9.0.5/go.mod h1:WqMKv5vnQbRuZstUwxQI195wHy+t4PuXDOjzMvcuQHk=
+github.com/redis/go-redis/v9 v9.2.1 h1:WlYJg71ODF0dVspZZCpYmoF1+U1Jjk9Rwd7pq6QmlCg=
+github.com/redis/go-redis/v9 v9.2.1/go.mod h1:hdY0cQFCN4fnSYT6TkisLufl/4W5UIXyv0b/CLO2V2M=
 github.com/rivo/uniseg v0.2.0/go.mod h1:J6wj4VEh+S6ZtnVlnTBMWIodfgj8LQOQFoIToxlJtxc=
 github.com/rivo/uniseg v0.4.4 h1:8TfxU8dW6PdqD27gjM8MVNuicgxIjxpm4K7x4jp8sis=
 github.com/rivo/uniseg v0.4.4/go.mod h1:FN3SvrM+Zdj16jyLfmOkMNblXMcoc8DfTHruCPUcx88=