Generic NodeID implementation v2

async.go

@@ -31,27 +31,27 @@ func (f *Async[Resp]) Done() bool {
 // AsyncMajority returns an Async future that resolves when a majority quorum is reached.
 // Messages are sent immediately (synchronously) to preserve ordering when multiple
 // async calls are created in sequence.
-func (r *Responses[Resp]) AsyncMajority() *Async[Resp] {
+func (r *Responses[T, Resp]) AsyncMajority() *Async[Resp] {
 	quorumSize := r.size/2 + 1
 	return r.AsyncThreshold(quorumSize)
 }
 
 // AsyncFirst returns an Async future that resolves when the first response is received.
 // Messages are sent immediately (synchronously) to preserve ordering.
-func (r *Responses[Resp]) AsyncFirst() *Async[Resp] {
+func (r *Responses[T, Resp]) AsyncFirst() *Async[Resp] {
 	return r.AsyncThreshold(1)
 }
 
 // AsyncAll returns an Async future that resolves when all nodes have responded.
 // Messages are sent immediately (synchronously) to preserve ordering.
-func (r *Responses[Resp]) AsyncAll() *Async[Resp] {
+func (r *Responses[T, Resp]) AsyncAll() *Async[Resp] {
 	return r.AsyncThreshold(r.size)
 }
 
 // AsyncThreshold returns an Async future that resolves when the threshold is reached.
 // Messages are sent immediately (synchronously) to preserve ordering when multiple
 // async calls are created in sequence.
-func (r *Responses[Resp]) AsyncThreshold(threshold int) *Async[Resp] {
+func (r *Responses[T, Resp]) AsyncThreshold(threshold int) *Async[Resp] {
 	// Send messages synchronously before spawning the goroutine to preserve ordering
 	r.sendNow()
 

async_test.go

@@ -13,7 +13,7 @@ import (
 
 func TestAsync(t *testing.T) {
 	// a type alias short hand for the responses type
-	type respType = *gorums.Responses[*pb.StringValue]
+	type respType = *gorums.Responses[uint32, *pb.StringValue]
 	tests := []struct {
 		name      string
 		call      func(respType) *gorums.Async[*pb.StringValue]
@@ -45,7 +45,7 @@ func TestAsync(t *testing.T) {
 		t.Run(tt.name, func(t *testing.T) {
 			config := gorums.TestConfiguration(t, tt.numNodes, gorums.EchoServerFn)
 			ctx := gorums.TestContext(t, 2*time.Second)
-			responses := gorums.QuorumCall[*pb.StringValue, *pb.StringValue](
+			responses := gorums.QuorumCall[uint32, *pb.StringValue, *pb.StringValue](
 				config.Context(ctx),
 				pb.String("test"),
 				mock.TestMethod,
@@ -71,7 +71,7 @@ func TestAsync_Error(t *testing.T) {
 	ctx, cancel := context.WithCancel(context.Background())
 	cancel() // Cancel immediately
 
-	responses := gorums.QuorumCall[*pb.StringValue, *pb.StringValue](
+	responses := gorums.QuorumCall[uint32, *pb.StringValue, *pb.StringValue](
 		config.Context(ctx),
 		pb.String("test"),
 		mock.TestMethod,
@@ -92,7 +92,7 @@ func BenchmarkAsyncQuorumCall(b *testing.B) {
 		b.Run(fmt.Sprintf("AsyncMajority/%d", numNodes), func(b *testing.B) {
 			b.ReportAllocs()
 			for b.Loop() {
-				future := gorums.QuorumCall[*pb.StringValue, *pb.StringValue](
+				future := gorums.QuorumCall[uint32, *pb.StringValue, *pb.StringValue](
 					cfgCtx,
 					pb.String("test"),
 					mock.TestMethod,
@@ -108,7 +108,7 @@ func BenchmarkAsyncQuorumCall(b *testing.B) {
 		b.Run(fmt.Sprintf("BlockingMajority/%d", numNodes), func(b *testing.B) {
 			b.ReportAllocs()
 			for b.Loop() {
-				_, err := gorums.QuorumCall[*pb.StringValue, *pb.StringValue](
+				_, err := gorums.QuorumCall[uint32, *pb.StringValue, *pb.StringValue](
 					cfgCtx,
 					pb.String("test"),
 					mock.TestMethod,

benchmark/benchmark.go

@@ -35,8 +35,8 @@ type Bench struct {
 
 type (
 	benchFunc   func(Options) (*Result, error)
-	qcFunc      func(*gorums.ConfigContext, *Echo, int, ...gorums.CallOption) (*Echo, error)
-	asyncQCFunc func(*gorums.ConfigContext, *Echo, int, ...gorums.CallOption) AsyncEcho
+	qcFunc      func(*ConfigContext, *Echo, int, ...gorums.CallOption) (*Echo, error)
+	asyncQCFunc func(*ConfigContext, *Echo, int, ...gorums.CallOption) AsyncEcho
 	serverFunc  func(context.Context, *TimedMsg)
 )
 
@@ -233,7 +233,7 @@ func GetBenchmarks(config Configuration) []Bench {
 			Name:        "QuorumCall",
 			Description: "NodeStream based quorum call implementation with FIFO ordering",
 			runBench: func(opts Options) (*Result, error) {
-				return runQCBenchmark(opts, config, func(ctx *gorums.ConfigContext, in *Echo, quorumSize int, callOpts ...gorums.CallOption) (*Echo, error) {
+				return runQCBenchmark(opts, config, func(ctx *ConfigContext, in *Echo, quorumSize int, callOpts ...gorums.CallOption) (*Echo, error) {
 					return QuorumCall(ctx, in, callOpts...).Threshold(quorumSize)
 				})
 			},
@@ -242,7 +242,7 @@ func GetBenchmarks(config Configuration) []Bench {
 			Name:        "AsyncQuorumCall",
 			Description: "NodeStream based async quorum call implementation with FIFO ordering",
 			runBench: func(opts Options) (*Result, error) {
-				return runAsyncQCBenchmark(opts, config, func(ctx *gorums.ConfigContext, in *Echo, quorumSize int, callOpts ...gorums.CallOption) AsyncEcho {
+				return runAsyncQCBenchmark(opts, config, func(ctx *ConfigContext, in *Echo, quorumSize int, callOpts ...gorums.CallOption) AsyncEcho {
 					return QuorumCall(ctx, in, callOpts...).AsyncThreshold(quorumSize)
 				})
 			},
@@ -251,7 +251,7 @@ func GetBenchmarks(config Configuration) []Bench {
 			Name:        "SlowServer",
 			Description: "Quorum Call with a 10s processing time on the server",
 			runBench: func(opts Options) (*Result, error) {
-				return runQCBenchmark(opts, config, func(ctx *gorums.ConfigContext, in *Echo, quorumSize int, callOpts ...gorums.CallOption) (*Echo, error) {
+				return runQCBenchmark(opts, config, func(ctx *ConfigContext, in *Echo, quorumSize int, callOpts ...gorums.CallOption) (*Echo, error) {
 					return SlowServer(ctx, in, callOpts...).Threshold(quorumSize)
 				})
 			},

callopts.go

@@ -1,7 +1,6 @@
 package gorums
 
 import (
-	"google.golang.org/protobuf/proto"
 	"google.golang.org/protobuf/runtime/protoimpl"
 )
 
@@ -54,7 +53,7 @@ func IgnoreErrors() CallOption {
 //	resp, err := ReadQC(ctx, req,
 //	    gorums.Interceptors(loggingInterceptor, filterInterceptor),
 //	).Majority()
-func Interceptors[Req, Resp proto.Message](interceptors ...QuorumInterceptor[Req, Resp]) CallOption {
+func Interceptors[T NodeID, Req, Resp msg](interceptors ...QuorumInterceptor[T, Req, Resp]) CallOption {
 	return func(o *callOptions) {
 		for _, interceptor := range interceptors {
 			o.interceptors = append(o.interceptors, interceptor)

callopts_test.go

@@ -33,7 +33,9 @@ func TestCallOptionsMustWaitSendDone(t *testing.T) {
 }
 
 func BenchmarkGetCallOptions(b *testing.B) {
-	interceptor := func(_ *ClientCtx[msg, msg], next ResponseSeq[msg]) ResponseSeq[msg] { return next }
+	interceptor := func(_ *ClientCtx[uint32, msg, msg], next ResponseSeq[uint32, msg]) ResponseSeq[uint32, msg] {
+		return next
+	}
 	tests := []struct {
 		numOpts int
 	}{