Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions channel_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -177,7 +177,7 @@ func TestChannelSendBufferSize(t *testing.T) {
node := TestNode(t, EchoServerFn, WithSendBufferSize(size))

ctx := TestContext(t, time.Second)
_, err := RPCCall(node.Context(ctx), pb.String("test"), mock.TestMethod)
_, err := RPCCall[*pb.StringValue, *pb.StringValue](node.Context(ctx), pb.String("test"), mock.TestMethod)
if err != nil {
t.Errorf("RPCCall failed with buffer size %d: %v", size, err)
}
Expand All @@ -199,7 +199,7 @@ func TestChannelLatency(t *testing.T) {
// exponential weighted moving average to stabilize to the real RTT.
for i := range 10 {
ctx := TestContext(t, time.Second)
_, err := RPCCall(node.Context(ctx), pb.String("ping"), mock.TestMethod)
_, err := RPCCall[*pb.StringValue, *pb.StringValue](node.Context(ctx), pb.String("ping"), mock.TestMethod)
if err != nil {
t.Fatalf("RPCCall %d failed: %v", i, err)
}
Expand Down
6 changes: 1 addition & 5 deletions cmd/protoc-gen-gorums/dev/zorums_rpc_gorums.pb.go

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

15 changes: 5 additions & 10 deletions cmd/protoc-gen-gorums/gengorums/template_rpc.go
Original file line number Diff line number Diff line change
Expand Up @@ -9,22 +9,17 @@ var rpcComment = `
{{end -}}
`

var rpcSignature = `func {{$method}}(` +
`ctx *{{$nodeContext}}, in *{{$in}}) (resp *{{$out}}, err error) {
`

var rpcVar = `
{{$genFile := .GenFile}}
{{$nodeContext := "NodeContext"}}
{{$rpcCall := use "gorums.RPCCall" .GenFile}}
{{$rpc := use "gorums.RPCCall" .GenFile}}
{{$_ := use "gorums.EnforceVersion" .GenFile}}
`

var rpcBody = ` res, err := {{$rpcCall}}(ctx, in, "{{$fullName}}")
if err != nil {
return nil, err
}
return res.(*{{$out}}), err
var rpcSignature = `func {{$method}}(ctx *{{$nodeContext}}, in *{{$in}}) (*{{$out}}, error) {
`

var rpcBody = ` return {{$rpc}}[*{{$in}}, *{{$out}}](ctx, in, "{{$fullName}}")
}
`

Expand Down
2 changes: 1 addition & 1 deletion config_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -441,7 +441,7 @@ func TestConfigConcurrentAccess(t *testing.T) {
var wg sync.WaitGroup
for range 2 {
wg.Go(func() {
_, err := gorums.RPCCall(node.Context(t.Context()), pb.String(""), mock.TestMethod)
_, err := gorums.RPCCall[*pb.StringValue, *pb.StringValue](node.Context(t.Context()), pb.String(""), mock.TestMethod)
if err != nil {
errCh <- err
}
Expand Down
12 changes: 2 additions & 10 deletions examples/storage/proto/storage_gorums.pb.go

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

12 changes: 2 additions & 10 deletions internal/tests/metadata/metadata_gorums.pb.go

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

6 changes: 1 addition & 5 deletions internal/tests/ordering/order_gorums.pb.go

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

6 changes: 1 addition & 5 deletions internal/tests/tls/tls_gorums.pb.go

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

6 changes: 1 addition & 5 deletions internal/tests/unresponsive/unresponsive_gorums.pb.go

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

19 changes: 14 additions & 5 deletions rpc.go
Original file line number Diff line number Diff line change
Expand Up @@ -8,15 +8,24 @@ import (
// RPCCall executes a remote procedure call on the node.
//
// This method should be used by generated code only.
func RPCCall(ctx *NodeContext, msg proto.Message, method string) (proto.Message, error) {
func RPCCall[Req, Resp proto.Message](ctx *NodeContext, req Req, method string) (Resp, error) {
md := ordering.NewGorumsMetadata(ctx, ctx.nextMsgID(), method)
replyChan := make(chan NodeResponse[proto.Message], 1)
ctx.enqueue(request{ctx: ctx, msg: NewRequestMessage(md, msg), responseChan: replyChan})
replyChan := make(chan NodeResponse[msg], 1)
ctx.enqueue(request{ctx: ctx, msg: NewRequestMessage(md, req), responseChan: replyChan})

select {
case r := <-replyChan:
return r.Value, r.Err
var zero Resp
if r.Err != nil {
return zero, r.Err
}
resp, ok := r.Value.(Resp)
if !ok {
return zero, ErrTypeMismatch
}
return resp, r.Err
case <-ctx.Done():
return nil, ctx.Err()
var zero Resp
return zero, ctx.Err()
}
}
20 changes: 17 additions & 3 deletions rpc_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,7 @@ func TestRPCCallSuccess(t *testing.T) {

ctx := gorums.TestContext(t, 5*time.Second)
nodeCtx := node.Context(ctx)
response, err := gorums.RPCCall(nodeCtx, pb.String(""), mock.TestMethod)
response, err := gorums.RPCCall[*pb.StringValue, *pb.StringValue](nodeCtx, pb.String(""), mock.TestMethod)
if err != nil {
t.Fatalf("Unexpected error, got: %v, want: %v", err, nil)
}
Expand All @@ -40,7 +40,7 @@ func TestRPCCallDownedNode(t *testing.T) {

ctx := gorums.TestContext(t, 5*time.Second)
nodeCtx := node.Context(ctx)
response, err := gorums.RPCCall(nodeCtx, pb.String(""), mock.TestMethod)
response, err := gorums.RPCCall[*pb.StringValue, *pb.StringValue](nodeCtx, pb.String(""), mock.TestMethod)
if err == nil {
t.Fatalf("Expected error, got: %v, want: %v", err, fmt.Errorf("rpc error: code = Unavailable desc = stream is down"))
}
Expand All @@ -56,11 +56,25 @@ func TestRPCCallTimedOut(t *testing.T) {
time.Sleep(50 * time.Millisecond)
defer cancel()
nodeCtx := node.Context(ctx)
response, err := gorums.RPCCall(nodeCtx, pb.String(""), mock.TestMethod)
response, err := gorums.RPCCall[*pb.StringValue, *pb.StringValue](nodeCtx, pb.String(""), mock.TestMethod)
if err == nil {
t.Fatalf("Expected error, got: %v, want: %v", err, fmt.Errorf("context deadline exceeded"))
}
if response != nil {
t.Fatalf("Unexpected response, got: %v, want: %v", response, nil)
}
}

func TestRPCCallTypeMismatch(t *testing.T) {
node := gorums.TestNode(t, gorums.DefaultTestServer)

ctx := gorums.TestContext(t, 5*time.Second)
nodeCtx := node.Context(ctx)
response, err := gorums.RPCCall[*pb.StringValue, *pb.Int32Value](nodeCtx, pb.String(""), mock.TestMethod)
if err != gorums.ErrTypeMismatch {
t.Fatalf("Expected error, got: %v, want: %v", err, gorums.ErrTypeMismatch)
}
if response != nil {
t.Fatalf("Unexpected response, got: %v, want: %v", response, nil)
}
}
8 changes: 4 additions & 4 deletions server_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -89,7 +89,7 @@ func TestServerInterceptorsChain(t *testing.T) {

ctx := gorums.TestContext(t, 5*time.Second)
nodeCtx := node.Context(ctx)
res, err := gorums.RPCCall(nodeCtx, pb.String("client-"), mock.TestMethod)
res, err := gorums.RPCCall[*pb.StringValue, *pb.StringValue](nodeCtx, pb.String("client-"), mock.TestMethod)
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
Expand Down Expand Up @@ -132,7 +132,7 @@ func TestTCPReconnection(t *testing.T) {
// Send first message
ctx := gorums.TestContext(t, time.Second)
nodeCtx := node.Context(ctx)
_, err = gorums.RPCCall(nodeCtx, pb.String("1"), mock.TestMethod)
_, err = gorums.RPCCall[*pb.StringValue, *pb.StringValue](nodeCtx, pb.String("1"), mock.TestMethod)
if err != nil {
t.Fatalf("First call failed: %v", err)
}
Expand All @@ -147,7 +147,7 @@ func TestTCPReconnection(t *testing.T) {
// Sending now should fail or timeout
ctx2 := gorums.TestContext(t, 200*time.Millisecond)
nodeCtx2 := node.Context(ctx2)
_, err = gorums.RPCCall(nodeCtx2, pb.String("2"), mock.TestMethod)
_, err = gorums.RPCCall[*pb.StringValue, *pb.StringValue](nodeCtx2, pb.String("2"), mock.TestMethod)
if err == nil {
// It might succeed if it just queued it? But we wait for response.
} else {
Expand Down Expand Up @@ -175,7 +175,7 @@ func TestTCPReconnection(t *testing.T) {
// Send message again
ctx3 := gorums.TestContext(t, 2*time.Second)
nodeCtx3 := node.Context(ctx3)
_, err = gorums.RPCCall(nodeCtx3, pb.String("3"), mock.TestMethod)
_, err = gorums.RPCCall[*pb.StringValue, *pb.StringValue](nodeCtx3, pb.String("3"), mock.TestMethod)
if err != nil {
t.Errorf("Call after reconnection failed: %v", err)
}
Expand Down