matrix-capnproto-fed

commit b5e7d6c76a85f0e31aa32ba890430499f2b0e001
parent da4877611cb1f565a7c1357109b96558a6c06cda
Author: MTRNord <mtrnord1@gmail.com>
Date:   Thu, 14 Mar 2024 21:57:51 +0100

Switch to packed (and compressed with lz4) encoding as well as adding missing Release calls on the client

Diffstat:
M
cmds/client/main.go
 | 
6
+++++-
M
cmds/server/main.go
 | 
58
+++++++++++++++++++++++++++++++++++++++++++++++++++++++---

2 files changed, 60 insertions(+), 4 deletions(-)
diff --git a/cmds/client/main.go b/cmds/client/main.go
@@ -6,6 +6,7 @@ import (
 	"net"
 	"time"
 
+	"capnproto.org/go/capnp/v3/flowcontrol/bbr"
 	"capnproto.org/go/capnp/v3/rpc"
 	"github.com/MTRNord/matrix_protobuf_fed/cmds/client/helpers"
 	protocol "github.com/MTRNord/matrix_protobuf_fed/proto/federation/v1"
@@ -25,12 +26,13 @@ func main() {
 	log.Println("Connected to server on port localhost:2000")
 	log.Println("Creating rpc client...")
 
-	rpc_conn := rpc.NewConn(rpc.NewStreamTransport(conn), nil)
+	rpc_conn := rpc.NewConn(rpc.NewPackedStreamTransport(conn), nil)
 	defer rpc_conn.Close()
 
 	log.Println("Created rpc connection...")
 
 	client := protocol.MatrixFederation(rpc_conn.Bootstrap(context.TODO()))
+	defer client.Release()
 
 	log.Println("Created client...")
 
@@ -52,6 +54,8 @@ func main() {
 func printServerKeys(ctx context.Context, client protocol.MatrixFederation) error {
 	callback := helpers.NewKeyStreamCallback()
 	callback_client := protocol.StreamCallback_ServerToClient(callback)
+	defer callback_client.Release()
+	callback_client.SetFlowLimiter(bbr.NewLimiter(nil))
 	keys_future, release := client.GetKeys(ctx, func(p protocol.MatrixFederation_getKeys_Params) error {
 		log.Println("Sending getKeys request...")
 
diff --git a/cmds/server/main.go b/cmds/server/main.go
@@ -2,10 +2,12 @@ package main
 
 import (
 	"context"
+	"errors"
 	"log"
+	"net"
 
 	"capnproto.org/go/capnp/v3"
-	"capnproto.org/go/capnp/v3/flowcontrol"
+	"capnproto.org/go/capnp/v3/flowcontrol/bbr"
 	"capnproto.org/go/capnp/v3/rpc"
 	"github.com/MTRNord/matrix_protobuf_fed/rpcserver"
 
@@ -14,12 +16,62 @@ import (
 
 // This implements a dummy server for testing purposes of the rough api design especially around signatures
 
+// Serve serves a Cap'n Proto RPC to incoming connections.
+//
+// Serve will take ownership of bootstrapClient and release it after the listener closes.
+//
+// Serve exits with the listener error if the listener is closed by the owner.
+func Serve(lis net.Listener, boot capnp.Client) error {
+	if !boot.IsValid() {
+		err := errors.New("bootstrap client is not valid")
+		return err
+	}
+	// Since we took ownership of the bootstrap client, release it after we're done.
+	defer boot.Release()
+	for {
+		// Accept incoming connections
+		conn, err := lis.Accept()
+		if err != nil {
+			return err
+		}
+
+		// the RPC connection takes ownership of the bootstrap interface and will release it when the connection
+		// exits, so use AddRef to avoid releasing the provided bootstrap client capability.
+		opts := rpc.Options{
+			BootstrapClient: boot.AddRef(),
+		}
+		// For each new incoming connection, create a new RPC transport connection that will serve incoming RPC requests
+		transport := rpc.NewPackedStreamTransport(conn)
+		_ = rpc.NewConn(transport, &opts)
+	}
+}
+
+// ListenAndServe opens a listener on the given address and serves a Cap'n Proto RPC to incoming connections
+//
+// network and address are passed to net.Listen. Use network "unix" for Unix Domain Sockets
+// and "tcp" for regular TCP IP4 or IP6 connections.
+//
+// ListenAndServe will take ownership of bootstrapClient and release it on exit.
+func ListenAndServe(ctx context.Context, network, addr string, bootstrapClient capnp.Client) error {
+	listener, err := net.Listen(network, addr)
+
+	if err == nil {
+		// to close this listener, close the context
+		go func() {
+			<-ctx.Done()
+			_ = listener.Close()
+		}()
+		err = Serve(listener, bootstrapClient)
+	}
+	return err
+}
+
 func main() {
 	log.Println("Starting server on port localhost:2000")
 	server := rpcserver.NewServer()
 
 	client := protocol.MatrixFederation_ServerToClient(server)
-	client.SetFlowLimiter(flowcontrol.NewFixedLimiter(1 << 16))
+	client.SetFlowLimiter(bbr.NewLimiter(nil))
 
-	rpc.ListenAndServe(context.Background(), "tcp", "localhost:2000", capnp.Client(client))
+	ListenAndServe(context.Background(), "tcp", "localhost:2000", capnp.Client(client))
 }