Function consumer_group_round_robin

nodedb/src/control/pubsub.rs:441–471

()

    #[tokio::test]
    async fn consumer_group_round_robin() {
        let registry = TopicRegistry::new(1000);
        registry.create_topic("tasks").unwrap();

        // Two members in the same consumer group.
        let (_id1, mut rx1, _) = registry.subscribe_group("tasks", "workers", 0).unwrap();
        let (_id2, mut rx2, _) = registry.subscribe_group("tasks", "workers", 0).unwrap();

        // Publish 4 messages.
        registry.publish("tasks", "a".into(), "system").unwrap();
        registry.publish("tasks", "b".into(), "system").unwrap();
        registry.publish("tasks", "c".into(), "system").unwrap();
        registry.publish("tasks", "d".into(), "system").unwrap();

        // Each member should get ~2 messages (round-robin).
        let mut count1 = 0;
        let mut count2 = 0;
        while let Ok(msg) = rx1.try_recv() {
            count1 += 1;
            let _ = msg;
        }
        while let Ok(msg) = rx2.try_recv() {
            count2 += 1;
            let _ = msg;
        }
        assert_eq!(count1 + count2, 4);
        assert!(
            count1 >= 1 && count2 >= 1,
            "both members should get messages: c1={count1}, c2={count2}"
        );
    }

    #[test]
    fn consumer_group_stats() {