test: add custom aggregation pattern tests for Responses

Author: meling

responses_test.go

@@ -303,3 +303,140 @@ func TestIteratorMethods(t *testing.T) {
 		}
 	})
 }
+
+// -------------------------------------------------------------------------
+// Custom Aggregation Pattern Tests
+// -------------------------------------------------------------------------
+
+// TestCustomAggregation demonstrates how users can define custom aggregation
+// functions that operate on *Responses and return custom types.
+func TestCustomAggregation(t *testing.T) {
+	t.Run("SameTypeAggregation", func(t *testing.T) {
+		// Aggregation function that returns the same type (Resp -> Resp)
+		majorityQF := func(resp *Responses[*pb.StringValue]) (*pb.StringValue, error) {
+			replies := resp.IgnoreErrors().CollectN(2)
+			if len(replies) < 2 {
+				return nil, ErrIncomplete
+			}
+			for _, v := range replies {
+				return v, nil
+			}
+			return nil, ErrIncomplete
+		}
+
+		responses := []NodeResponse[proto.Message]{
+			{NodeID: 1, Value: pb.String("response1"), Err: nil},
+			{NodeID: 2, Value: pb.String("response2"), Err: nil},
+			{NodeID: 3, Value: pb.String("response3"), Err: nil},
+		}
+		clientCtx := makeClientCtx[*pb.StringValue, *pb.StringValue](t, 3, responses)
+		r := NewResponses(clientCtx)
+
+		// Call the aggregation function directly
+		result, err := majorityQF(r)
+		if err != nil {
+			t.Fatalf("Expected no error, got %v", err)
+		}
+		if result.GetValue() != "response1" && result.GetValue() != "response2" {
+			t.Errorf("Expected response1 or response2, got %s", result.GetValue())
+		}
+	})
+
+	t.Run("CustomReturnType", func(t *testing.T) {
+		// Aggregation function that returns a different type (Resp -> []string)
+		// This demonstrates the key benefit: Out can differ from In
+		collectAllValues := func(resp *Responses[*pb.StringValue]) ([]string, error) {
+			replies := resp.IgnoreErrors().CollectAll()
+			if len(replies) == 0 {
+				return nil, ErrIncomplete
+			}
+			result := make([]string, 0, len(replies))
+			for _, v := range replies {
+				result = append(result, v.GetValue())
+			}
+			return result, nil
+		}
+
+		responses := []NodeResponse[proto.Message]{
+			{NodeID: 1, Value: pb.String("alpha"), Err: nil},
+			{NodeID: 2, Value: pb.String("beta"), Err: nil},
+			{NodeID: 3, Value: pb.String("gamma"), Err: nil},
+		}
+		clientCtx := makeClientCtx[*pb.StringValue, *pb.StringValue](t, 3, responses)
+		r := NewResponses(clientCtx)
+
+		// Call the aggregation function directly - returns []string from *Responses[*pb.StringValue]
+		result, err := collectAllValues(r)
+		if err != nil {
+			t.Fatalf("Expected no error, got %v", err)
+		}
+		if len(result) != 3 {
+			t.Errorf("Expected 3 values, got %d", len(result))
+		}
+	})
+
+	t.Run("WithFiltering", func(t *testing.T) {
+		// Aggregation function that uses filtering and custom logic
+		filterAndCount := func(resp *Responses[*pb.StringValue]) (int, error) {
+			count := 0
+			for range resp.IgnoreErrors().Filter(func(r NodeResponse[*pb.StringValue]) bool {
+				return r.NodeID > 1 // Only nodes 2 and 3
+			}) {
+				count++
+			}
+			if count == 0 {
+				return 0, ErrIncomplete
+			}
+			return count, nil
+		}
+
+		responses := []NodeResponse[proto.Message]{
+			{NodeID: 1, Value: pb.String("response1"), Err: nil},
+			{NodeID: 2, Value: pb.String("response2"), Err: nil},
+			{NodeID: 3, Value: pb.String("response3"), Err: nil},
+		}
+		clientCtx := makeClientCtx[*pb.StringValue, *pb.StringValue](t, 3, responses)
+		r := NewResponses(clientCtx)
+
+		// Call the aggregation function directly
+		count, err := filterAndCount(r)
+		if err != nil {
+			t.Fatalf("Expected no error, got %v", err)
+		}
+		if count != 2 {
+			t.Errorf("Expected 2 filtered responses, got %d", count)
+		}
+	})
+
+	t.Run("ErrorHandling", func(t *testing.T) {
+		// Aggregation function that handles errors explicitly
+		requireAllSuccess := func(resp *Responses[*pb.StringValue]) (*pb.StringValue, error) {
+			var first *pb.StringValue
+			for r := range resp.Seq() {
+				if r.Err != nil {
+					return nil, r.Err
+				}
+				if first == nil {
+					first = r.Value
+				}
+			}
+			if first == nil {
+				return nil, ErrIncomplete
+			}
+			return first, nil
+		}
+
+		responses := []NodeResponse[proto.Message]{
+			{NodeID: 1, Value: pb.String("response1"), Err: nil},
+			{NodeID: 2, Value: nil, Err: errors.New("node 2 failed")},
+		}
+		clientCtx := makeClientCtx[*pb.StringValue, *pb.StringValue](t, 2, responses)
+		r := NewResponses(clientCtx)
+
+		// Call the aggregation function directly
+		_, err := requireAllSuccess(r)
+		if err == nil {
+			t.Error("Expected error, got nil")
+		}
+	})
+}