[pitboss] phase 20: Track M.2 — MessageHandler end-to-end (Kafka / SQS / Pub-Sub / NATS / RabbitMQ)

src/dynamic/stubs/broker_kafka.rs

@@ -0,0 +1,109 @@
+//! Phase 20 (Track M.2) — Kafka broker loopback stub source-snippet provider.
+//!
+//! The Phase 20 acceptance gate runs every per-lang `MessageHandler` harness
+//! inside an in-process loopback broker — no real Kafka cluster, no
+//! external network — so the per-lang harness can publish the spec's
+//! payload onto a topic and observe the handler under test receive it
+//! synchronously.  Each `broker_kafka` source snippet declares a tiny
+//! `NyxKafkaLoopback` type whose `publish(topic, payload)` immediately
+//! routes the bytes through the subscriber callback the harness has
+//! registered.  No threads, no sockets, no async runtime: a single
+//! synchronous in-process dispatch keeps Phase 10's 500 ms boot budget
+//! intact when `stubs_required` is empty.
+//!
+//! The snippet shape mirrors [`crate::dynamic::stubs::mocks::mock_source`] —
+//! per-language inline source returned as a `&'static str` so the
+//! generated harness can splice it verbatim into its own source.  The
+//! per-language harness emitter is responsible for instantiating the
+//! loopback and invoking the registered handler with the payload.
+
+use crate::symbol::Lang;
+
+/// Marker text the loopback emits on stdout when the harness publishes
+/// a message.  Stable across languages so a future
+/// `ProbeKind::BrokerPublish` predicate can pin the byte sequence.
+pub const KAFKA_PUBLISH_MARKER: &str = "__NYX_BROKER_PUBLISH__:kafka";
+
+/// Source snippet declaring an in-process Kafka loopback for `lang`.
+/// Returns `""` when the language has no harness-level Kafka adapter
+/// (everything outside Java / Python today).  The snippet does *not*
+/// emit a publish marker by itself; the per-lang harness emitter calls
+/// `publish(topic, payload)` and prints the marker once.
+pub fn kafka_source(lang: Lang) -> &'static str {
+    match lang {
+        Lang::Python => {
+            r#"
+class NyxKafkaLoopback:
+    """In-process Kafka loopback — no socket, no thread, no broker."""
+    def __init__(self):
+        self._subs = {}
+    def subscribe(self, topic, cb):
+        self._subs.setdefault(topic, []).append(cb)
+    def publish(self, topic, payload):
+        for cb in self._subs.get(topic, []):
+            cb(payload)
+"#
+        }
+        Lang::Java => {
+            r#"
+    static class NyxKafkaLoopback {
+        private final java.util.Map<String, java.util.List<java.util.function.Consumer<String>>> subs = new java.util.HashMap<>();
+        public void subscribe(String topic, java.util.function.Consumer<String> cb) {
+            subs.computeIfAbsent(topic, k -> new java.util.ArrayList<>()).add(cb);
+        }
+        public void publish(String topic, String payload) {
+            for (java.util.function.Consumer<String> cb : subs.getOrDefault(topic, java.util.Collections.emptyList())) {
+                cb.accept(payload);
+            }
+        }
+    }
+"#
+        }
+        _ => "",
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn kafka_publish_marker_is_stable() {
+        assert_eq!(KAFKA_PUBLISH_MARKER, "__NYX_BROKER_PUBLISH__:kafka");
+    }
+
+    #[test]
+    fn python_snippet_declares_loopback_class() {
+        let src = kafka_source(Lang::Python);
+        assert!(src.contains("class NyxKafkaLoopback"));
+        assert!(src.contains("def publish"));
+        assert!(src.contains("def subscribe"));
+    }
+
+    #[test]
+    fn java_snippet_declares_static_inner_class() {
+        let src = kafka_source(Lang::Java);
+        assert!(src.contains("static class NyxKafkaLoopback"));
+        assert!(src.contains("public void publish"));
+        assert!(src.contains("public void subscribe"));
+    }
+
+    #[test]
+    fn unsupported_langs_return_empty_snippet() {
+        for lang in [
+            Lang::Go,
+            Lang::JavaScript,
+            Lang::TypeScript,
+            Lang::Php,
+            Lang::Ruby,
+            Lang::Rust,
+            Lang::C,
+            Lang::Cpp,
+        ] {
+            assert!(
+                kafka_source(lang).is_empty(),
+                "{lang:?} should not yet ship a Kafka loopback snippet"
+            );
+        }
+    }
+}

src/dynamic/stubs/broker_nats.rs

@@ -0,0 +1,81 @@
+//! Phase 20 (Track M.2) — NATS broker loopback stub.
+//!
+//! Mints `nats.io/nats.go` style `*nats.Msg` envelopes (`Subject`,
+//! `Data`, `Reply`) for Go handlers.
+
+use crate::symbol::Lang;
+
+/// Stdout sentinel printed once per publish.
+pub const NATS_PUBLISH_MARKER: &str = "__NYX_BROKER_PUBLISH__:nats";
+
+/// Source snippet declaring an in-process NATS loopback for `lang`.
+pub fn nats_source(lang: Lang) -> &'static str {
+    match lang {
+        Lang::Go => {
+            r#"
+type NyxNatsMsg struct {
+    Subject string
+    Data    []byte
+    Reply   string
+}
+
+type NyxNatsLoopback struct {
+    subs map[string][]func(*NyxNatsMsg)
+}
+
+func NewNyxNatsLoopback() *NyxNatsLoopback {
+    return &NyxNatsLoopback{subs: map[string][]func(*NyxNatsMsg){}}
+}
+
+func (l *NyxNatsLoopback) Subscribe(subject string, cb func(*NyxNatsMsg)) {
+    l.subs[subject] = append(l.subs[subject], cb)
+}
+
+func (l *NyxNatsLoopback) Publish(subject string, payload string) {
+    msg := &NyxNatsMsg{Subject: subject, Data: []byte(payload)}
+    for _, cb := range l.subs[subject] {
+        cb(msg)
+    }
+}
+"#
+        }
+        _ => "",
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn marker_stable() {
+        assert_eq!(NATS_PUBLISH_MARKER, "__NYX_BROKER_PUBLISH__:nats");
+    }
+
+    #[test]
+    fn go_loopback_exposes_subject_data_reply() {
+        let src = nats_source(Lang::Go);
+        assert!(src.contains("type NyxNatsMsg struct"));
+        assert!(src.contains("Subject string"));
+        assert!(src.contains("Data    []byte"));
+        assert!(src.contains("Reply   string"));
+        assert!(src.contains("func NewNyxNatsLoopback"));
+    }
+
+    #[test]
+    fn other_langs_return_empty_snippet() {
+        for lang in [
+            Lang::Python,
+            Lang::Java,
+            Lang::JavaScript,
+            Lang::TypeScript,
+            Lang::Php,
+            Lang::Ruby,
+            Lang::Rust,
+            Lang::C,
+            Lang::Cpp,
+        ] {
+            assert!(nats_source(lang).is_empty());
+        }
+    }
+}

src/dynamic/stubs/broker_pubsub.rs

@@ -0,0 +1,100 @@
+//! Phase 20 (Track M.2) — Google Pub/Sub broker loopback stub.
+//!
+//! Mints `google.cloud.pubsub_v1.subscriber.message.Message`-shaped
+//! envelopes (`message_id`, `data`, `ack`, `nack`) for Python / Go.
+
+use crate::symbol::Lang;
+
+/// Stdout sentinel the per-lang harness prints once per publish.
+pub const PUBSUB_PUBLISH_MARKER: &str = "__NYX_BROKER_PUBLISH__:pubsub";
+
+/// Source snippet declaring an in-process Pub/Sub loopback for `lang`.
+pub fn pubsub_source(lang: Lang) -> &'static str {
+    match lang {
+        Lang::Python => {
+            r#"
+class NyxPubsubMessage:
+    def __init__(self, mid, data):
+        self.message_id = mid
+        self.data = data if isinstance(data, (bytes, bytearray)) else data.encode('utf-8', 'replace')
+        self.acked = False
+        self.nacked = False
+    def ack(self): self.acked = True
+    def nack(self): self.nacked = True
+
+class NyxPubsubLoopback:
+    def __init__(self):
+        self._subs = {}
+        self._mid = 0
+    def subscribe(self, topic, cb):
+        self._subs.setdefault(topic, []).append(cb)
+    def publish(self, topic, payload):
+        self._mid += 1
+        msg = NyxPubsubMessage(f'nyx-{self._mid:08d}', payload)
+        for cb in self._subs.get(topic, []):
+            cb(msg)
+"#
+        }
+        Lang::Go => {
+            r#"
+type NyxPubsubMessage struct {
+    ID    string
+    Data  []byte
+    Acked bool
+}
+
+func (m *NyxPubsubMessage) Ack()  { m.Acked = true }
+func (m *NyxPubsubMessage) Nack() { m.Acked = false }
+
+type NyxPubsubLoopback struct {
+    subs map[string][]func(*NyxPubsubMessage)
+    mid  int
+}
+
+func NewNyxPubsubLoopback() *NyxPubsubLoopback {
+    return &NyxPubsubLoopback{subs: map[string][]func(*NyxPubsubMessage){}}
+}
+
+func (l *NyxPubsubLoopback) Subscribe(topic string, cb func(*NyxPubsubMessage)) {
+    l.subs[topic] = append(l.subs[topic], cb)
+}
+
+func (l *NyxPubsubLoopback) Publish(topic string, payload string) {
+    l.mid += 1
+    msg := &NyxPubsubMessage{ID: fmt.Sprintf("nyx-%08d", l.mid), Data: []byte(payload)}
+    for _, cb := range l.subs[topic] {
+        cb(msg)
+    }
+}
+"#
+        }
+        _ => "",
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn marker_stable() {
+        assert_eq!(PUBSUB_PUBLISH_MARKER, "__NYX_BROKER_PUBLISH__:pubsub");
+    }
+
+    #[test]
+    fn python_carries_ack_nack_surface() {
+        let src = pubsub_source(Lang::Python);
+        assert!(src.contains("class NyxPubsubMessage"));
+        assert!(src.contains("def ack"));
+        assert!(src.contains("def nack"));
+        assert!(src.contains("message_id"));
+    }
+
+    #[test]
+    fn go_carries_ack_nack_methods() {
+        let src = pubsub_source(Lang::Go);
+        assert!(src.contains("type NyxPubsubMessage struct"));
+        assert!(src.contains("func (m *NyxPubsubMessage) Ack"));
+        assert!(src.contains("NewNyxPubsubLoopback"));
+    }
+}

src/dynamic/stubs/broker_rabbit.rs

@@ -0,0 +1,88 @@
+//! Phase 20 (Track M.2) — RabbitMQ broker loopback stub.
+//!
+//! Mints `pika.BasicProperties` / `com.rabbitmq.client.Envelope`-shaped
+//! envelopes for Python / Java handlers.
+
+use crate::symbol::Lang;
+
+/// Stdout sentinel printed once per publish.
+pub const RABBIT_PUBLISH_MARKER: &str = "__NYX_BROKER_PUBLISH__:rabbit";
+
+/// Source snippet declaring an in-process RabbitMQ loopback for `lang`.
+pub fn rabbit_source(lang: Lang) -> &'static str {
+    match lang {
+        Lang::Python => {
+            r#"
+class NyxRabbitProperties:
+    def __init__(self, mid):
+        self.message_id = mid
+        self.delivery_mode = 2
+
+class NyxRabbitMethod:
+    def __init__(self, tag, routing_key):
+        self.delivery_tag = tag
+        self.routing_key = routing_key
+
+class NyxRabbitChannel:
+    def __init__(self):
+        self._subs = {}
+        self._tag = 0
+    def basic_consume(self, queue, on_message_callback, **kw):
+        self._subs.setdefault(queue, []).append(on_message_callback)
+    def basic_publish(self, exchange, routing_key, body, properties=None):
+        self._tag += 1
+        method = NyxRabbitMethod(self._tag, routing_key)
+        props = properties or NyxRabbitProperties(f'nyx-{self._tag:08d}')
+        body_bytes = body if isinstance(body, (bytes, bytearray)) else body.encode('utf-8', 'replace')
+        for cb in self._subs.get(routing_key, []):
+            cb(self, method, props, body_bytes)
+"#
+        }
+        Lang::Java => {
+            r#"
+    static class NyxRabbitChannel {
+        private final java.util.Map<String, java.util.List<java.util.function.BiConsumer<String, String>>> subs = new java.util.HashMap<>();
+        private long tag = 0;
+        public void basicConsume(String queue, java.util.function.BiConsumer<String, String> cb) {
+            subs.computeIfAbsent(queue, k -> new java.util.ArrayList<>()).add(cb);
+        }
+        public void basicPublish(String exchange, String routingKey, String body) {
+            tag += 1;
+            String mid = "nyx-" + tag;
+            for (java.util.function.BiConsumer<String, String> cb : subs.getOrDefault(routingKey, java.util.Collections.emptyList())) {
+                cb.accept(mid, body);
+            }
+        }
+    }
+"#
+        }
+        _ => "",
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn marker_stable() {
+        assert_eq!(RABBIT_PUBLISH_MARKER, "__NYX_BROKER_PUBLISH__:rabbit");
+    }
+
+    #[test]
+    fn python_carries_pika_shape() {
+        let src = rabbit_source(Lang::Python);
+        assert!(src.contains("class NyxRabbitChannel"));
+        assert!(src.contains("basic_consume"));
+        assert!(src.contains("basic_publish"));
+        assert!(src.contains("delivery_tag"));
+    }
+
+    #[test]
+    fn java_carries_static_inner_channel() {
+        let src = rabbit_source(Lang::Java);
+        assert!(src.contains("static class NyxRabbitChannel"));
+        assert!(src.contains("basicConsume"));
+        assert!(src.contains("basicPublish"));
+    }
+}

src/dynamic/stubs/broker_sqs.rs

@@ -0,0 +1,119 @@
+//! Phase 20 (Track M.2) — SQS broker loopback stub source-snippet provider.
+//!
+//! Mirrors [`crate::dynamic::stubs::broker_kafka`] but mints SQS-shaped
+//! envelopes (`MessageId`, `ReceiptHandle`, `Body`) the way `boto3.sqs` /
+//! `software.amazon.awssdk.services.sqs` / the AWS Node SDK present
+//! them.  The loopback never speaks the AWS protocol — it just calls
+//! the registered handler synchronously with a single-message envelope.
+
+use crate::symbol::Lang;
+
+/// Stdout sentinel the per-lang harness prints once per publish.
+pub const SQS_PUBLISH_MARKER: &str = "__NYX_BROKER_PUBLISH__:sqs";
+
+/// Source snippet declaring an in-process SQS loopback for `lang`.
+/// Java / Python / Node (JS+TS) carry concrete snippets; every other
+/// lang returns `""`.
+pub fn sqs_source(lang: Lang) -> &'static str {
+    match lang {
+        Lang::Python => {
+            r#"
+class NyxSqsLoopback:
+    """In-process SQS loopback — boto3-shaped envelopes."""
+    def __init__(self):
+        self._subs = {}
+        self._mid = 0
+    def subscribe(self, queue, cb):
+        self._subs.setdefault(queue, []).append(cb)
+    def publish(self, queue, payload):
+        self._mid += 1
+        envelope = {
+            'MessageId': f'nyx-{self._mid:08d}',
+            'ReceiptHandle': f'rh-nyx-{self._mid:08d}',
+            'Body': payload,
+        }
+        for cb in self._subs.get(queue, []):
+            cb(envelope)
+"#
+        }
+        Lang::Java => {
+            r#"
+    static class NyxSqsLoopback {
+        private final java.util.Map<String, java.util.List<java.util.function.Consumer<java.util.Map<String, String>>>> subs = new java.util.HashMap<>();
+        private int mid = 0;
+        public void subscribe(String queue, java.util.function.Consumer<java.util.Map<String, String>> cb) {
+            subs.computeIfAbsent(queue, k -> new java.util.ArrayList<>()).add(cb);
+        }
+        public void publish(String queue, String payload) {
+            mid += 1;
+            java.util.Map<String, String> envelope = new java.util.HashMap<>();
+            envelope.put("MessageId", "nyx-" + mid);
+            envelope.put("ReceiptHandle", "rh-nyx-" + mid);
+            envelope.put("Body", payload);
+            for (java.util.function.Consumer<java.util.Map<String, String>> cb : subs.getOrDefault(queue, java.util.Collections.emptyList())) {
+                cb.accept(envelope);
+            }
+        }
+    }
+"#
+        }
+        Lang::JavaScript | Lang::TypeScript => {
+            r#"
+class NyxSqsLoopback {
+    constructor() { this._subs = new Map(); this._mid = 0; }
+    subscribe(queue, cb) {
+        if (!this._subs.has(queue)) this._subs.set(queue, []);
+        this._subs.get(queue).push(cb);
+    }
+    publish(queue, payload) {
+        this._mid += 1;
+        const envelope = {
+            MessageId: 'nyx-' + this._mid,
+            ReceiptHandle: 'rh-nyx-' + this._mid,
+            Body: payload,
+        };
+        for (const cb of (this._subs.get(queue) || [])) cb(envelope);
+    }
+}
+"#
+        }
+        _ => "",
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn marker_stable() {
+        assert_eq!(SQS_PUBLISH_MARKER, "__NYX_BROKER_PUBLISH__:sqs");
+    }
+
+    #[test]
+    fn python_carries_boto3_shape() {
+        let src = sqs_source(Lang::Python);
+        assert!(src.contains("class NyxSqsLoopback"));
+        assert!(src.contains("MessageId"));
+        assert!(src.contains("ReceiptHandle"));
+        assert!(src.contains("Body"));
+    }
+
+    #[test]
+    fn java_carries_envelope_map() {
+        let src = sqs_source(Lang::Java);
+        assert!(src.contains("static class NyxSqsLoopback"));
+        assert!(src.contains("MessageId"));
+        assert!(src.contains("Body"));
+    }
+
+    #[test]
+    fn node_class_supports_subscribe_publish() {
+        let src = sqs_source(Lang::JavaScript);
+        assert!(src.contains("class NyxSqsLoopback"));
+        assert!(src.contains("subscribe(queue"));
+        assert!(src.contains("publish(queue"));
+        let ts = sqs_source(Lang::TypeScript);
+        assert_eq!(ts, src);
+    }
+}

src/dynamic/stubs/mod.rs

@@ -51,6 +51,11 @@
 //!   [`crate::dynamic::oracle::oracle_fired_with_stubs`] so the
 //!   `StubEventMatches` predicate can satisfy a payload.
 
+pub mod broker_kafka;
+pub mod broker_nats;
+pub mod broker_pubsub;
+pub mod broker_rabbit;
+pub mod broker_sqs;
 pub mod filesystem;
 pub mod http;
 pub mod ldap_server;
@@ -59,6 +64,11 @@ pub mod redis;
 pub mod sql;
 pub mod xpath_document;
 
+pub use broker_kafka::{kafka_source, KAFKA_PUBLISH_MARKER};
+pub use broker_nats::{nats_source, NATS_PUBLISH_MARKER};
+pub use broker_pubsub::{pubsub_source, PUBSUB_PUBLISH_MARKER};
+pub use broker_rabbit::{rabbit_source, RABBIT_PUBLISH_MARKER};
+pub use broker_sqs::{sqs_source, SQS_PUBLISH_MARKER};
 pub use filesystem::FilesystemStub;
 pub use http::HttpStub;
 pub use ldap_server::LdapStub;