diff --git a/CLAUDE.md b/CLAUDE.md index 167d4fe..f9fe1ab 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -109,14 +109,15 @@ Built on top of NATS (transport layer), mq9 adds store-first semantics and prior ``` $mq9.AI.MAILBOX.CREATE PUB+reply create mailbox -$mq9.AI.MAILBOX.MSG.{mail_id}.{priority} PUB send message -$mq9.AI.MAILBOX.MSG.{mail_id}.* SUB subscribe (all priorities) -$mq9.AI.MAILBOX.MSG.{mail_id}.{priority} SUB subscribe (one priority) +$mq9.AI.MAILBOX.MSG.{mail_id} PUB/SUB send/receive default (normal) message +$mq9.AI.MAILBOX.MSG.{mail_id}.{priority} PUB send urgent or critical message +$mq9.AI.MAILBOX.MSG.{mail_id}.* SUB subscribe urgent+critical (excludes default normal) +$mq9.AI.MAILBOX.MSG.{mail_id}.{priority} SUB subscribe one non-default priority $mq9.AI.MAILBOX.LIST.{mail_id} PUB+reply list messages snapshot (no payload) $mq9.AI.MAILBOX.DELETE.{mail_id}.{msg_id} PUB+reply delete one message ``` -Priority values: `high` / `normal` / `low` +Priority values: `critical` (highest) / `urgent` / `normal` (default, no suffix) ### Key semantics @@ -229,7 +230,7 @@ Each language has a standalone project that imports the published SDK v0.3.5. All demos run the same scenario: 1. Connect to `nats://localhost:4222` 2. Create a private mailbox (TTL 60s) -3. Send 3 messages (high / normal / low priority) +3. Send 3 messages (critical / urgent / normal priority) 4. Subscribe and print received messages 5. List mailbox metadata, delete one message 6. Create a public mailbox diff --git a/README.md b/README.md index beab69c..e5e30d7 100644 --- a/README.md +++ b/README.md @@ -12,7 +12,7 @@ mq9 gives every agent a durable mailbox. Messages persist until TTL expires — | Concept | Description | |---------|-------------| | **Mailbox** | Agent's address. Private (UUID) or public (user-defined name). TTL-driven, auto-cleaned. | -| **Priority** | `high` / `normal` / `low`. Cross-priority ordering guaranteed by storage. | +| **Priority** | `critical` / `urgent` / `normal` (default, no suffix). Cross-priority ordering guaranteed by storage. | | **Store-first** | Subscriber gets all non-expired messages on connect, then real-time going forward. | | **Queue group** | Multiple subscribers sharing a group receive each message exactly once. | @@ -21,7 +21,8 @@ mq9 gives every agent a durable mailbox. Messages persist until TTL expires — | Operation | Subject | |-----------|---------| | Create mailbox | `$mq9.AI.MAILBOX.CREATE` | -| Send message | `$mq9.AI.MAILBOX.MSG.{mail_id}.{priority}` | +| Send message (default) | `$mq9.AI.MAILBOX.MSG.{mail_id}` | +| Send message (urgent/critical) | `$mq9.AI.MAILBOX.MSG.{mail_id}.{priority}` | | Subscribe | `$mq9.AI.MAILBOX.MSG.{mail_id}.*` | | List metadata | `$mq9.AI.MAILBOX.LIST.{mail_id}` | | Delete message | `$mq9.AI.MAILBOX.DELETE.{mail_id}.{msg_id}` | @@ -34,12 +35,12 @@ Full spec: [docs/mq9-protocol.md](docs/mq9-protocol.md) | Language | Package | Version | Install | |----------|---------|---------|---------| -| Python | `robustmq` | 0.3.5 | `pip install robustmq` | -| Go | `github.com/robustmq/robustmq-sdk/go` | v0.3.5 | `go get github.com/robustmq/robustmq-sdk/go` | -| JavaScript | `@robustmq/sdk` | 0.3.5 | `npm install @robustmq/sdk` | -| Java | `com.robustmq:robustmq` | 0.3.5 | Maven / Gradle (see below) | -| Rust | `robustmq` | 0.3.5 | `cargo add robustmq` | -| C# | `RobustMQ` | 0.3.5 | `dotnet add package RobustMQ` | +| Python | `robustmq` | 1.0.1 | `pip install robustmq` | +| Go | `github.com/robustmq/robustmq-sdk/go` | v1.0.1 | `go get github.com/robustmq/robustmq-sdk/go` | +| JavaScript | `@robustmq/sdk` | 1.0.1 | `npm install @robustmq/sdk` | +| Java | `com.robustmq:robustmq` | 1.0.1 | Maven / Gradle (see below) | +| Rust | `robustmq` | 1.0.1 | `cargo add robustmq` | +| C# | `RobustMQ` | 1.0.1 | `dotnet add package RobustMQ` | ### Integrations @@ -99,7 +100,7 @@ await client.send(mailbox.mailId, "hello", "normal"); com.robustmq robustmq - 0.3.5 + 1.0.1 ``` ```java @@ -113,7 +114,7 @@ client.send(mailbox.getMailId(), "hello".getBytes(), Priority.NORMAL).get(); **Java (Gradle)** ```groovy -implementation 'com.robustmq:robustmq:0.3.5' +implementation 'com.robustmq:robustmq:1.0.1' ``` **Rust** @@ -201,7 +202,7 @@ Full examples: [demo/demo-langchain-mq9/](demo/demo-langchain-mq9/) · [demo/dem Each demo is a standalone project that connects to `nats://demo.robustmq.com:4222` and runs the same scenario: 1. Create a private mailbox (TTL 60s) -2. Send 3 messages (high / normal / low priority) +2. Send 3 messages (critical / urgent / normal priority) 3. Subscribe and print received messages 4. List mailbox metadata, delete one message 5. Create a public mailbox @@ -261,7 +262,7 @@ demo/ demo-langchain-mq9/ # langchain-mq9 demo demo-multi-agent/ # cross-language multi-agent demo (Python + Go) demo-langgraph/ # LangGraph workflow demo -VERSION # Canonical version (currently 0.3.5) +VERSION # Canonical version (currently 1.0.1) ``` --- diff --git a/VERSION b/VERSION index c2c0004..7dea76e 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -0.3.5 +1.0.1 diff --git a/csharp/RobustMQ.Mq9.Tests/MQ9ClientTests.cs b/csharp/RobustMQ.Mq9.Tests/MQ9ClientTests.cs index 1845172..8a58fbd 100644 --- a/csharp/RobustMQ.Mq9.Tests/MQ9ClientTests.cs +++ b/csharp/RobustMQ.Mq9.Tests/MQ9ClientTests.cs @@ -92,41 +92,42 @@ public async Task Create_ServerError_ThrowsMQ9Error() // ── send ────────────────────────────────────────────────────────────────── [Fact] - public async Task Send_PublishesCorrectSubject() + public async Task Send_NormalPriority_UsesBareSubject() { var mock = Substitute.For(); var client = ClientWithMock(mock); await client.SendAsync("m-001", "hello"u8.ToArray(), Priority.Normal); + // normal uses bare subject (no suffix) await mock.Received(1).PublishAsync( - "$mq9.AI.MAILBOX.MSG.m-001.normal", + "$mq9.AI.MAILBOX.MSG.m-001", Arg.Any()); } [Fact] - public async Task Send_HighPriority_CorrectSubject() + public async Task Send_CriticalPriority_UsesSuffix() { var mock = Substitute.For(); var client = ClientWithMock(mock); - await client.SendAsync("m-001", "urgent"u8.ToArray(), Priority.High); + await client.SendAsync("m-001", "abort"u8.ToArray(), Priority.Critical); await mock.Received(1).PublishAsync( - "$mq9.AI.MAILBOX.MSG.m-001.high", + "$mq9.AI.MAILBOX.MSG.m-001.critical", Arg.Any()); } [Fact] - public async Task Send_LowPriority_CorrectSubject() + public async Task Send_UrgentPriority_UsesSuffix() { var mock = Substitute.For(); var client = ClientWithMock(mock); - await client.SendAsync("m-001", "bg"u8.ToArray(), Priority.Low); + await client.SendAsync("m-001", "interrupt"u8.ToArray(), Priority.Urgent); await mock.Received(1).PublishAsync( - "$mq9.AI.MAILBOX.MSG.m-001.low", + "$mq9.AI.MAILBOX.MSG.m-001.urgent", Arg.Any()); } @@ -151,7 +152,7 @@ public async Task Subscribe_AllPriorities_UsesWildcardSubject() } [Fact] - public async Task Subscribe_SinglePriority_UsesSpecificSubject() + public async Task Subscribe_CriticalPriority_UsesSuffixedSubject() { var mock = Substitute.For(); var capturedSubject = ""; @@ -164,9 +165,29 @@ public async Task Subscribe_SinglePriority_UsesSpecificSubject() var client = ClientWithMock(mock); await using var _ = await client.SubscribeAsync("m-001", async _ => await Task.CompletedTask, - priority: Priority.High); + priority: Priority.Critical); - Assert.Equal("$mq9.AI.MAILBOX.MSG.m-001.high", capturedSubject); + Assert.Equal("$mq9.AI.MAILBOX.MSG.m-001.critical", capturedSubject); + } + + [Fact] + public async Task Subscribe_NormalPriority_UsesBareSubject() + { + var mock = Substitute.For(); + var capturedSubject = ""; + mock.SubscribeCoreAsync(Arg.Any(), cancellationToken: Arg.Any()) + .Returns(callInfo => + { + capturedSubject = callInfo.ArgAt(0); + return ValueTask.FromResult(Substitute.For>()); + }); + var client = ClientWithMock(mock); + + await using var _ = await client.SubscribeAsync("m-001", async _ => await Task.CompletedTask, + priority: Priority.Normal); + + // normal uses bare subject (no suffix) + Assert.Equal("$mq9.AI.MAILBOX.MSG.m-001", capturedSubject); } [Fact] @@ -210,7 +231,7 @@ await Assert.ThrowsAsync( public async Task List_ReturnsMessageMeta() { var json = "{\"mail_id\":\"m-001\",\"messages\":[{\"msg_id\":\"x1\"," - + "\"priority\":\"high\",\"ts\":100}]}"; + + "\"priority\":\"critical\",\"ts\":100}]}"; var mock = Substitute.For(); mock.RequestAsync( @@ -223,7 +244,7 @@ public async Task List_ReturnsMessageMeta() Assert.Single(metas); Assert.Equal("x1", metas[0].MsgId); - Assert.Equal(Priority.High, metas[0].Priority); + Assert.Equal(Priority.Critical, metas[0].Priority); Assert.Equal(100L, metas[0].Ts); } diff --git a/csharp/RobustMQ.Mq9/MQ9Client.cs b/csharp/RobustMQ.Mq9/MQ9Client.cs index fbbf14d..d87df1b 100644 --- a/csharp/RobustMQ.Mq9/MQ9Client.cs +++ b/csharp/RobustMQ.Mq9/MQ9Client.cs @@ -82,7 +82,9 @@ public async Task SendAsync(string mailId, byte[] payload, Priority priority = Priority.Normal) { EnsureConnected(); - var subject = $"{Prefix}.MAILBOX.MSG.{mailId}.{priority.ToSubjectString()}"; + var subject = priority == Priority.Normal + ? $"{Prefix}.MAILBOX.MSG.{mailId}" + : $"{Prefix}.MAILBOX.MSG.{mailId}.{priority.ToSubjectString()}"; await _nc!.PublishAsync(subject, payload); } @@ -96,7 +98,9 @@ public async Task SubscribeAsync( { EnsureConnected(); var subject = priority.HasValue - ? $"{Prefix}.MAILBOX.MSG.{mailId}.{priority.Value.ToSubjectString()}" + ? (priority.Value == Priority.Normal + ? $"{Prefix}.MAILBOX.MSG.{mailId}" + : $"{Prefix}.MAILBOX.MSG.{mailId}.{priority.Value.ToSubjectString()}") : $"{Prefix}.MAILBOX.MSG.{mailId}.*"; var sub = string.IsNullOrEmpty(queueGroup) @@ -182,9 +186,11 @@ private async Task RequestAsync(string subject, JsonNode payload) private static Mq9Message ParseIncoming(string mailId, string subject, byte[] data) { - // Extract priority from last subject token - var parts = subject.Split('.'); - var priorityStr = parts.Length > 0 ? parts[^1] : "normal"; + // Subject is bare $mq9.AI.MAILBOX.MSG.{mailId} (normal) + // or $mq9.AI.MAILBOX.MSG.{mailId}.{urgent|critical} + var msgBase = $"{Prefix}.MAILBOX.MSG.{mailId}"; + var suffix = subject.StartsWith(msgBase) ? subject[msgBase.Length..] : ""; + var priorityStr = suffix.StartsWith(".") ? suffix[1..] : ""; var priority = PriorityExtensions.FromString(priorityStr); try diff --git a/csharp/RobustMQ.Mq9/Models.cs b/csharp/RobustMQ.Mq9/Models.cs index b859ea6..f76fec8 100644 --- a/csharp/RobustMQ.Mq9/Models.cs +++ b/csharp/RobustMQ.Mq9/Models.cs @@ -2,25 +2,24 @@ namespace RobustMQ.Mq9; public enum Priority { - High, - Normal, - Low + Critical, + Urgent, + Normal } internal static class PriorityExtensions { public static string ToSubjectString(this Priority p) => p switch { - Priority.High => "high", - Priority.Normal => "normal", - Priority.Low => "low", + Priority.Critical => "critical", + Priority.Urgent => "urgent", _ => "normal" }; public static Priority FromString(string s) => s switch { - "high" => Priority.High, - "low" => Priority.Low, + "critical" => Priority.Critical, + "urgent" => Priority.Urgent, _ => Priority.Normal }; } diff --git a/csharp/RobustMQ.Mq9/RobustMQ.Mq9.csproj b/csharp/RobustMQ.Mq9/RobustMQ.Mq9.csproj index 7ab4635..b278916 100644 --- a/csharp/RobustMQ.Mq9/RobustMQ.Mq9.csproj +++ b/csharp/RobustMQ.Mq9/RobustMQ.Mq9.csproj @@ -7,7 +7,7 @@ RobustMQ RobustMQ.Mq9 RobustMQ - 0.3.5 + 1.0.1 RobustMQ SDK — unified messaging for AI, IoT, and data pipelines Apache-2.0 https://github.com/robustmq/robustmq-sdk diff --git a/demo/demo-csharp/Program.cs b/demo/demo-csharp/Program.cs index d64bcfd..10c7bd4 100644 --- a/demo/demo-csharp/Program.cs +++ b/demo/demo-csharp/Program.cs @@ -19,10 +19,10 @@ Console.WriteLine($"[csharp] private mailbox: {mailbox.MailId}"); // 2. Send 3 messages with different priorities -await client.SendAsync(mailbox.MailId, "urgent task"u8.ToArray(), Priority.High); +await client.SendAsync(mailbox.MailId, "abort signal"u8.ToArray(), Priority.Critical); +await client.SendAsync(mailbox.MailId, "urgent task"u8.ToArray(), Priority.Urgent); await client.SendAsync(mailbox.MailId, "normal task"u8.ToArray(), Priority.Normal); -await client.SendAsync(mailbox.MailId, "background task"u8.ToArray(), Priority.Low); -Console.WriteLine("[csharp] sent 3 messages (high / normal / low)"); +Console.WriteLine("[csharp] sent 3 messages (critical / urgent / normal)"); // 3. Subscribe and print received messages await using var sub = await client.SubscribeAsync(mailbox.MailId, async msg => diff --git a/demo/demo-csharp/demo-csharp.csproj b/demo/demo-csharp/demo-csharp.csproj index e61b430..a95a55a 100644 --- a/demo/demo-csharp/demo-csharp.csproj +++ b/demo/demo-csharp/demo-csharp.csproj @@ -8,7 +8,7 @@ - + diff --git a/demo/demo-go/main.go b/demo/demo-go/main.go index a669f33..6ae9a60 100644 --- a/demo/demo-go/main.go +++ b/demo/demo-go/main.go @@ -33,10 +33,10 @@ func main() { fmt.Printf("[go] private mailbox: %s\n", mailbox.MailID) // 2. Send 3 messages with different priorities - _ = c.Send(mailbox.MailID, []byte("urgent task"), mq9.High) + _ = c.Send(mailbox.MailID, []byte("abort signal"), mq9.Critical) + _ = c.Send(mailbox.MailID, []byte("urgent task"), mq9.Urgent) _ = c.Send(mailbox.MailID, []byte("normal task"), mq9.Normal) - _ = c.Send(mailbox.MailID, []byte("background task"), mq9.Low) - fmt.Println("[go] sent 3 messages (high / normal / low)") + fmt.Println("[go] sent 3 messages (critical / urgent / normal)") // 3. Subscribe and print received messages sub, err := c.Subscribe(mailbox.MailID, func(msg *mq9.Message) { diff --git a/demo/demo-java/pom.xml b/demo/demo-java/pom.xml index 6da7d8a..811ba5c 100644 --- a/demo/demo-java/pom.xml +++ b/demo/demo-java/pom.xml @@ -7,7 +7,7 @@ io.robustmq.demo demo-java - 0.3.5 + 1.0.1 jar RobustMQ mq9 Java Demo @@ -22,7 +22,7 @@ com.robustmq robustmq - 0.3.5 + 1.0.1 diff --git a/demo/demo-java/src/main/java/com/robustmq/demo/Demo.java b/demo/demo-java/src/main/java/com/robustmq/demo/Demo.java index 86519b6..5619be9 100644 --- a/demo/demo-java/src/main/java/com/robustmq/demo/Demo.java +++ b/demo/demo-java/src/main/java/com/robustmq/demo/Demo.java @@ -27,10 +27,10 @@ public static void main(String[] args) throws Exception { System.out.println("[java] private mailbox: " + mailbox.getMailId()); // 2. Send 3 messages with different priorities - client.send(mailbox.getMailId(), "urgent task".getBytes(), Priority.HIGH).get(); + client.send(mailbox.getMailId(), "abort signal".getBytes(), Priority.CRITICAL).get(); + client.send(mailbox.getMailId(), "urgent task".getBytes(), Priority.URGENT).get(); client.send(mailbox.getMailId(), "normal task".getBytes(), Priority.NORMAL).get(); - client.send(mailbox.getMailId(), "background task".getBytes(), Priority.LOW).get(); - System.out.println("[java] sent 3 messages (high / normal / low)"); + System.out.println("[java] sent 3 messages (critical / urgent / normal)"); // 3. Subscribe and print received messages var dispatcher = client.subscribe(mailbox.getMailId(), msg -> diff --git a/demo/demo-javascript/demo.js b/demo/demo-javascript/demo.js index 66faad7..db10192 100644 --- a/demo/demo-javascript/demo.js +++ b/demo/demo-javascript/demo.js @@ -20,10 +20,10 @@ async function main() { console.log(`[js] private mailbox: ${mailbox.mailId}`); // 2. Send 3 messages with different priorities - await client.send(mailbox.mailId, "urgent task", "high"); + await client.send(mailbox.mailId, "abort signal", "critical"); + await client.send(mailbox.mailId, "urgent task", "urgent"); await client.send(mailbox.mailId, "normal task", "normal"); - await client.send(mailbox.mailId, "background task", "low"); - console.log("[js] sent 3 messages (high / normal / low)"); + console.log("[js] sent 3 messages (critical / urgent / normal)"); // 3. Subscribe and print received messages const sub = await client.subscribe(mailbox.mailId, async (msg) => { diff --git a/demo/demo-javascript/package.json b/demo/demo-javascript/package.json index 1a620cd..d8c4a16 100644 --- a/demo/demo-javascript/package.json +++ b/demo/demo-javascript/package.json @@ -8,6 +8,6 @@ "start": "node demo.js" }, "dependencies": { - "@robustmq/sdk": "0.3.5" + "@robustmq/sdk": "1.0.1" } } diff --git a/demo/demo-langchain-mq9/demo.py b/demo/demo-langchain-mq9/demo.py index 3841e25..4165cf4 100644 --- a/demo/demo-langchain-mq9/demo.py +++ b/demo/demo-langchain-mq9/demo.py @@ -4,7 +4,7 @@ Flow: Agent A creates a private mailbox Agent B creates a public mailbox (shared channel) - Agent B sends 3 messages (high / normal / low) to Agent A + Agent B sends 3 messages (critical / urgent / normal) to Agent A Agent A lists message metadata (no payload) Agent A reads messages with full content Agent A deletes the first message @@ -47,13 +47,13 @@ async def main() -> None: print(f"[agent-b] {result}") # ── Agent B: send 3 messages to Agent A ────────────────────────────────── - result = await send._arun(mail_id=mail_id, content="Urgent: pipeline failed!", priority="high") + result = await send._arun(mail_id=mail_id, content="ABORT: pipeline failed!", priority="critical") print(f"[agent-b] {result}") - result = await send._arun(mail_id=mail_id, content="Daily report is ready.", priority="normal") + result = await send._arun(mail_id=mail_id, content="Interrupt: daily report ready.", priority="urgent") print(f"[agent-b] {result}") - result = await send._arun(mail_id=mail_id, content="Background sync complete.", priority="low") + result = await send._arun(mail_id=mail_id, content="Background sync complete.", priority="normal") print(f"[agent-b] {result}") # ── Agent A: list message metadata (no payload) ─────────────────────────── diff --git a/demo/demo-langgraph/README.md b/demo/demo-langgraph/README.md index 4617fd0..da3f06f 100644 --- a/demo/demo-langgraph/README.md +++ b/demo/demo-langgraph/README.md @@ -74,7 +74,7 @@ FINAL: | Each LangGraph node owns a private mq9 mailbox | `node_writer` and `node_reviewer` each call `client.create()` | | Async handoff between nodes via mq9 | writer sends draft; reviewer subscribes, reads, replies | | Store-first semantics | reviewer can start after writer has already published | -| Priority-ordered delivery | draft sent as `NORMAL`, review sent back as `HIGH` | +| Priority-ordered delivery | draft sent as `NORMAL`, review sent back as `CRITICAL` | | LangChain LLM inside a LangGraph node | `ChatOpenAI(model="gpt-4o-mini")` called inside each node | ## Using `langchain-mq9` tools instead diff --git a/demo/demo-langgraph/langgraph_mq9_demo.py b/demo/demo-langgraph/langgraph_mq9_demo.py index e2c6448..5c19729 100644 --- a/demo/demo-langgraph/langgraph_mq9_demo.py +++ b/demo/demo-langgraph/langgraph_mq9_demo.py @@ -133,7 +133,7 @@ async def on_message(msg): print(f"[node_reviewer] review:\n{review}") # Send review back to writer - await client.send(reply_to, review.encode(), priority=Priority.HIGH) + await client.send(reply_to, review.encode(), priority=Priority.CRITICAL) print(f"[node_reviewer] sent review to writer ({reply_to})") return state diff --git a/demo/demo-multi-agent/README.md b/demo/demo-multi-agent/README.md index 334e348..15b1342 100644 --- a/demo/demo-multi-agent/README.md +++ b/demo/demo-multi-agent/README.md @@ -10,9 +10,9 @@ Agent A (Python) Agent B (Go) │ │ │── create private mailbox ──────────────│ │── publish reply address ──────────────▶│ (rendezvous mailbox) - │── send task [high] ──────────────────▶│ - │── send task [normal] ───────────────▶│ - │── send task [low] ──────────────────▶│ + │── send task [critical] ───────────────▶│ + │── send task [normal] ────────────────▶│ + │── send task [urgent] ───────────────▶│ │ │── process + send result ──▶ Agent A │ │── process + send result ──▶ Agent A │ │── process + send result ──▶ Agent A @@ -53,11 +53,11 @@ python agent_a.py ``` [agent-b] connected, waiting for messages on demo.multi-agent.rendezvous [agent-b] discovered Agent A reply address: m-550e8400-... -[agent-b] processing [high]: Analyze sentiment of Q3 earnings call +[agent-b] processing [critical]: Analyze sentiment of Q3 earnings call [agent-b] sent result to m-550e8400-...: DONE: Analyze sentiment of Q3 earnings call [agent-b] processing [normal]: Summarize the latest research papers [agent-b] sent result to m-550e8400-...: DONE: Summarize the latest research papers -[agent-b] processing [low]: Index new documents in the knowledge base +[agent-b] processing [urgent]: Index new documents in the knowledge base [agent-b] sent result to m-550e8400-...: DONE: Index new documents in the knowledge base [agent-b] done — processed 3 tasks ``` @@ -67,9 +67,9 @@ python agent_a.py ``` [agent-a] private mailbox: m-550e8400-... [agent-a] published reply address to 'demo.multi-agent.rendezvous' -[agent-a] sent task [high]: Analyze sentiment of Q3 earnings call +[agent-a] sent task [critical]: Analyze sentiment of Q3 earnings call [agent-a] sent task [normal]: Summarize the latest research papers -[agent-a] sent task [low]: Index new documents in the knowledge base +[agent-a] sent task [urgent]: Index new documents in the knowledge base [agent-a] received result [normal]: DONE: Analyze sentiment of Q3 earnings call [agent-a] received result [normal]: DONE: Summarize the latest research papers [agent-a] received result [normal]: DONE: Index new documents in the knowledge base @@ -82,6 +82,6 @@ python agent_a.py |---------|-------| | Private mailbox as a secure reply channel | `agent_a.py` — `client.create(ttl=120)` | | Public mailbox as an address rendezvous | `agent_a.py` — `client.create(..., public=True, name=rendezvous)` | -| Priority-ordered task delivery | Agent B receives `high` before `normal` before `low` | +| Priority-ordered task delivery | Agent B receives `critical` before `urgent` before `normal` | | Cross-language communication | Python SDK → Go SDK, same NATS subjects | | Store-first semantics | Agent B can start after Agent A publishes — it still gets all messages | diff --git a/demo/demo-multi-agent/agent_a.py b/demo/demo-multi-agent/agent_a.py index 5604b45..637a3b8 100644 --- a/demo/demo-multi-agent/agent_a.py +++ b/demo/demo-multi-agent/agent_a.py @@ -22,9 +22,9 @@ RENDEZVOUS = "demo.multi-agent.rendezvous" TASKS = [ - ("Analyze sentiment of Q3 earnings call", Priority.HIGH), - ("Summarize the latest research papers", Priority.NORMAL), - ("Index new documents in the knowledge base", Priority.LOW), + ("Analyze sentiment of Q3 earnings call", Priority.CRITICAL), + ("Summarize the latest research papers", Priority.URGENT), + ("Index new documents in the knowledge base", Priority.NORMAL), ] @@ -40,7 +40,7 @@ async def main() -> None: await client.send( RENDEZVOUS, mailbox.mail_id.encode(), - priority=Priority.HIGH, + priority=Priority.CRITICAL, ) print(f"[agent-a] published reply address to '{RENDEZVOUS}'") diff --git a/demo/demo-python/demo.py b/demo/demo-python/demo.py index 39a45dd..36d07a3 100644 --- a/demo/demo-python/demo.py +++ b/demo/demo-python/demo.py @@ -22,10 +22,10 @@ async def main() -> None: print(f"[python] private mailbox: {mailbox.mail_id}") # 2. Send 3 messages with different priorities - await client.send(mailbox.mail_id, b"urgent task", priority=Priority.HIGH) + await client.send(mailbox.mail_id, b"abort signal", priority=Priority.CRITICAL) + await client.send(mailbox.mail_id, b"urgent task", priority=Priority.URGENT) await client.send(mailbox.mail_id, b"normal task", priority=Priority.NORMAL) - await client.send(mailbox.mail_id, b"background task", priority=Priority.LOW) - print("[python] sent 3 messages (high / normal / low)") + print("[python] sent 3 messages (critical / urgent / normal)") # 3. Subscribe and print received messages async def handler(msg): diff --git a/demo/demo-python/pyproject.toml b/demo/demo-python/pyproject.toml index 3d39314..b0f4f0a 100644 --- a/demo/demo-python/pyproject.toml +++ b/demo/demo-python/pyproject.toml @@ -1,9 +1,9 @@ [project] name = "demo-python" -version = "0.3.5" +version = "1.0.1" requires-python = ">=3.10" dependencies = [ - "robustmq==0.3.5", + "robustmq==1.0.1", ] [build-system] diff --git a/demo/demo-python/requirements.txt b/demo/demo-python/requirements.txt index dc44c8e..b74eec3 100644 --- a/demo/demo-python/requirements.txt +++ b/demo/demo-python/requirements.txt @@ -1 +1 @@ -robustmq==0.3.5 +robustmq==1.0.1 diff --git a/demo/demo-rust/Cargo.toml b/demo/demo-rust/Cargo.toml index 23a42e9..533562d 100644 --- a/demo/demo-rust/Cargo.toml +++ b/demo/demo-rust/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "demo-rust" -version = "0.3.5" +version = "1.0.1" edition = "2021" [[bin]] @@ -8,5 +8,5 @@ name = "demo" path = "src/main.rs" [dependencies] -robustmq = { version = "0.3.5", features = ["mq9"] } +robustmq = { version = "1.0.1", features = ["mq9"] } tokio = { version = "1", features = ["full"] } diff --git a/demo/demo-rust/src/main.rs b/demo/demo-rust/src/main.rs index 966e633..2adfab6 100644 --- a/demo/demo-rust/src/main.rs +++ b/demo/demo-rust/src/main.rs @@ -18,10 +18,10 @@ async fn main() -> Result<(), Box> { println!("[rust] private mailbox: {}", mailbox.mail_id); // 2. Send 3 messages with different priorities - client.send(&mailbox.mail_id, b"urgent task", Priority::High).await?; + client.send(&mailbox.mail_id, b"abort signal", Priority::Critical).await?; + client.send(&mailbox.mail_id, b"urgent task", Priority::Urgent).await?; client.send(&mailbox.mail_id, b"normal task", Priority::Normal).await?; - client.send(&mailbox.mail_id, b"background task", Priority::Low).await?; - println!("[rust] sent 3 messages (high / normal / low)"); + println!("[rust] sent 3 messages (critical / urgent / normal)"); // 3. Subscribe and print received messages let _sub = client diff --git a/docs/csharp.md b/docs/csharp.md index 7efdae7..051a9d1 100644 --- a/docs/csharp.md +++ b/docs/csharp.md @@ -60,9 +60,9 @@ Task DeleteAsync(string mailId, string msgId) ## Priority ```csharp -Priority.High // "high" -Priority.Normal // "normal" -Priority.Low // "low" +Priority.Critical // "critical" — highest +Priority.Urgent // "urgent" +Priority.Normal // "normal" — default, no suffix ``` ## Public mailbox diff --git a/docs/go.md b/docs/go.md index 34b01d0..4c69c52 100644 --- a/docs/go.md +++ b/docs/go.md @@ -72,9 +72,9 @@ c.Delete(mailID, msgID string) error ## Priority ```go -mq9.High // "high" -mq9.Normal // "normal" -mq9.Low // "low" +mq9.Critical // "critical" — highest +mq9.Urgent // "urgent" +mq9.Normal // "normal" — default, no suffix ``` ## Public mailbox diff --git a/docs/java.md b/docs/java.md index d078547..059ae31 100644 --- a/docs/java.md +++ b/docs/java.md @@ -9,7 +9,7 @@ com.robustmq robustmq - 0.3.5 + 1.0.1 ``` @@ -70,9 +70,9 @@ CompletableFuture delete(String mailId, String msgId) ## Priority ```java -Priority.HIGH // "high" -Priority.NORMAL // "normal" -Priority.LOW // "low" +Priority.CRITICAL // "critical" — highest +Priority.URGENT // "urgent" +Priority.NORMAL // "normal" — default, no suffix ``` ## Public mailbox diff --git a/docs/javascript.md b/docs/javascript.md index 4679800..638fcc6 100644 --- a/docs/javascript.md +++ b/docs/javascript.md @@ -63,7 +63,7 @@ client.delete(mailId, msgId): Promise ## Priority ```typescript -"high" | "normal" | "low" +"critical" | "urgent" | "normal" // "normal" is default (no suffix in subject) ``` ## Public mailbox diff --git a/docs/langchain-mq9.md b/docs/langchain-mq9.md index 4d71278..2b39e10 100644 --- a/docs/langchain-mq9.md +++ b/docs/langchain-mq9.md @@ -92,14 +92,14 @@ Sends a message to any mailbox. The recipient does not need to be online — mq9 from langchain_mq9 import SendMessageTool tool = SendMessageTool(server="nats://demo.robustmq.com:4222") -tool.run({"mail_id": mail_id, "content": "Task complete.", "priority": "high"}) +tool.run({"mail_id": mail_id, "content": "Task complete.", "priority": "critical"}) ``` | Parameter | Type | Default | Description | |-----------|------|---------|-------------| | `mail_id` | str | — | Target mailbox address | | `content` | str | — | Message content | -| `priority` | str | `"normal"` | `"high"`, `"normal"`, or `"low"` | +| `priority` | str | `"normal"` | `"critical"`, `"urgent"`, or `"normal"` | --- @@ -113,7 +113,7 @@ from langchain_mq9 import GetMessagesTool tool = GetMessagesTool(server="nats://demo.robustmq.com:4222") result = tool.run({"mail_id": mail_id, "limit": 10}) # → "Found 2 message(s) in mailbox 'm-001': -# [1] priority=high content=Urgent: server down +# [1] priority=critical content=Urgent: server down # [2] priority=normal content=Daily report ready" ``` @@ -134,7 +134,7 @@ from langchain_mq9 import ListMessagesTool tool = ListMessagesTool(server="nats://demo.robustmq.com:4222") result = tool.run({"mail_id": mail_id}) # → "2 message(s) in mailbox 'm-001': -# msg_id=msg-001 priority=high ts=1234567890 +# msg_id=msg-001 priority=critical ts=1234567890 # msg_id=msg-002 priority=normal ts=1234567891" ``` @@ -192,7 +192,7 @@ async def main(): # Agent B: send messages await SendMessageTool(server=SERVER)._arun( - mail_id=mail_id, content="Urgent: data pipeline failed", priority="high" + mail_id=mail_id, content="Urgent: data pipeline failed", priority="critical" ) await SendMessageTool(server=SERVER)._arun( mail_id=mail_id, content="Weekly report attached", priority="normal" diff --git a/docs/mq9-protocol.md b/docs/mq9-protocol.md index 4fc18cd..cc003d0 100644 --- a/docs/mq9-protocol.md +++ b/docs/mq9-protocol.md @@ -29,9 +29,9 @@ Each message carries one of three priorities: | Value | Use | |-------|-----| -| `high` | Urgent (task abort, emergency commands) | -| `normal` | Regular communication (task distribution, results) | -| `low` | Background (logging, status reports) | +| `critical` | Highest priority — abort signals, emergency commands, security events | +| `urgent` | Urgent — task interrupts, time-sensitive instructions | +| `normal` (default, no suffix) | Routine communication — task dispatch, result delivery | - Same-priority messages: FIFO - Cross-priority: higher processed first @@ -97,29 +97,36 @@ For public mailboxes, `mail_id` equals the user-supplied `name`. --- -### MAILBOX.MSG.{mail_id}.{priority} +### MAILBOX.MSG.{mail_id} / MAILBOX.MSG.{mail_id}.{priority} -**Subject:** `$mq9.AI.MAILBOX.MSG.{mail_id}.{priority}` +**Subject (default/normal):** `$mq9.AI.MAILBOX.MSG.{mail_id}` (no suffix) +**Subject (urgent/critical):** `$mq9.AI.MAILBOX.MSG.{mail_id}.{priority}` **Direction:** PUB to send; SUB to receive **Sending a message:** -Publish to the subject where `priority` is one of `high`, `normal`, `low`. +Publish to the bare subject for `normal` (default), or append `.urgent` / `.critical` for higher priorities. ``` -PUB $mq9.AI.MAILBOX.MSG.m-uuid-001.normal +PUB $mq9.AI.MAILBOX.MSG.m-uuid-001 {"task": "summarize", "doc_id": "abc123"} + +PUB $mq9.AI.MAILBOX.MSG.m-uuid-001.urgent +{"type": "interrupt"} + +PUB $mq9.AI.MAILBOX.MSG.m-uuid-001.critical +{"type": "abort"} ``` Payload is arbitrary bytes — the server does not inspect it. **Receiving messages (subscribe):** -Subscribe with wildcard `*` to receive all priorities, or specify a single priority: - ``` -SUB $mq9.AI.MAILBOX.MSG.m-uuid-001.* # all priorities -SUB $mq9.AI.MAILBOX.MSG.m-uuid-001.high # high only +SUB $mq9.AI.MAILBOX.MSG.m-uuid-001.* # all priorities (normal, urgent, critical) +SUB $mq9.AI.MAILBOX.MSG.m-uuid-001 # normal only (bare subject, no suffix) +SUB $mq9.AI.MAILBOX.MSG.m-uuid-001.urgent # urgent only +SUB $mq9.AI.MAILBOX.MSG.m-uuid-001.critical # critical only ``` Subscription behavior: @@ -154,8 +161,8 @@ This is a metadata view — payloads are **not** included. { "mail_id": "m-uuid-001", "messages": [ - { "msg_id": "msg-001", "priority": "high", "ts": 1234567890 }, - { "msg_id": "msg-002", "priority": "normal", "ts": 1234567891 } + { "msg_id": "msg-001", "priority": "critical", "ts": 1234567890 }, + { "msg_id": "msg-002", "priority": "urgent", "ts": 1234567891 } ] } ``` @@ -208,8 +215,8 @@ Agents that need to communicate through a public mailbox agree on the name out-o # Subscribe to a known public mailbox nats sub '$mq9.AI.MAILBOX.MSG.task.queue.*' -# Send to a known public mailbox -nats pub '$mq9.AI.MAILBOX.MSG.task.queue.normal' 'payload' +# Send to a known public mailbox (normal/default priority, no suffix) +nats pub '$mq9.AI.MAILBOX.MSG.task.queue' 'payload' ``` Creating a public mailbox: @@ -284,16 +291,16 @@ nats req '$mq9.AI.MAILBOX.CREATE' '{"ttl":3600,"public":true,"name":"agent-a.inb # → {"mail_id":"agent-a.inbox"} # ── Step 3: Agent A publishes its private mail_id to the public mailbox ─────── -nats pub '$mq9.AI.MAILBOX.MSG.agent-a.inbox.normal' \ +nats pub '$mq9.AI.MAILBOX.MSG.agent-a.inbox' \ '{"reply_to":"m-550e8400-e29b-41d4-a716-446655440000"}' # ── Step 4: Agent B subscribes to the public mailbox to discover Agent A ────── -nats sub '$mq9.AI.MAILBOX.MSG.agent-a.inbox.*' +nats sub '$mq9.AI.MAILBOX.MSG.agent-a.inbox' # ← {"reply_to":"m-550e8400-e29b-41d4-a716-446655440000"} # Agent B now knows Agent A's private mail_id. # ── Step 5: Agent B sends a task result to Agent A's private mailbox ────────── -nats pub '$mq9.AI.MAILBOX.MSG.m-550e8400-e29b-41d4-a716-446655440000.high' \ +nats pub '$mq9.AI.MAILBOX.MSG.m-550e8400-e29b-41d4-a716-446655440000.critical' \ '{"status":"done","result":"analysis complete","score":0.97}' # ── Step 6: Agent A subscribes to its private mailbox and receives the result ─ @@ -303,7 +310,7 @@ nats sub '$mq9.AI.MAILBOX.MSG.m-550e8400-e29b-41d4-a716-446655440000.*' # ── Step 7: Agent A deletes the processed message ───────────────────────────── # First list to get msg_id: nats req '$mq9.AI.MAILBOX.LIST.m-550e8400-e29b-41d4-a716-446655440000' '{}' -# → {"mail_id":"m-550e...","messages":[{"msg_id":"msg-001","priority":"high","ts":1700000001}]} +# → {"mail_id":"m-550e...","messages":[{"msg_id":"msg-001","priority":"critical","ts":1700000001}]} nats req '$mq9.AI.MAILBOX.DELETE.m-550e8400-e29b-41d4-a716-446655440000.msg-001' '{}' # → {"deleted":true} @@ -336,20 +343,20 @@ nats sub '$mq9.AI.MAILBOX.MSG.task.queue.*' --queue workers # Each message is delivered to exactly one worker in the "workers" group. # ── Step 3: Producer sends 5 tasks with mixed priorities ────────────────────── -nats pub '$mq9.AI.MAILBOX.MSG.task.queue.high' '{"task_id":"t-001","type":"urgent-analysis"}' -nats pub '$mq9.AI.MAILBOX.MSG.task.queue.normal' '{"task_id":"t-002","type":"summarize"}' -nats pub '$mq9.AI.MAILBOX.MSG.task.queue.low' '{"task_id":"t-003","type":"index"}' -nats pub '$mq9.AI.MAILBOX.MSG.task.queue.high' '{"task_id":"t-004","type":"alert"}' -nats pub '$mq9.AI.MAILBOX.MSG.task.queue.normal' '{"task_id":"t-005","type":"report"}' +nats pub '$mq9.AI.MAILBOX.MSG.task.queue.critical' '{"task_id":"t-001","type":"urgent-analysis"}' +nats pub '$mq9.AI.MAILBOX.MSG.task.queue' '{"task_id":"t-002","type":"summarize"}' +nats pub '$mq9.AI.MAILBOX.MSG.task.queue.urgent' '{"task_id":"t-003","type":"interrupt"}' +nats pub '$mq9.AI.MAILBOX.MSG.task.queue.critical' '{"task_id":"t-004","type":"alert"}' +nats pub '$mq9.AI.MAILBOX.MSG.task.queue' '{"task_id":"t-005","type":"report"}' # ── Step 4: Observe delivery ─────────────────────────────────────────────────── -# High-priority tasks (t-001, t-004) are delivered before normal and low. +# Critical-priority tasks (t-001, t-004) are delivered before urgent and normal. # Each task goes to exactly one worker — no duplicate processing. # Workers that start after the producer still receive stored messages immediately. # Example worker output (distribution will vary): -# Worker 1 ← t-001 (high), t-003 (low) -# Worker 2 ← t-004 (high), t-005 (normal) +# Worker 1 ← t-001 (critical), t-003 (urgent) +# Worker 2 ← t-004 (critical), t-005 (normal) # Worker 3 ← t-002 (normal) # ── Step 5: Check remaining queue depth ─────────────────────────────────────── @@ -359,6 +366,6 @@ nats req '$mq9.AI.MAILBOX.LIST.task.queue' '{}' **Key points:** - All workers use the same `queue_group` name (`workers`). NATS delivers each message to exactly one member. -- High-priority messages stored before workers connect are still delivered in priority order on subscribe. +- Critical and urgent messages stored before workers connect are still delivered in priority order (critical → urgent → normal) on subscribe. - Workers can be added or removed at any time — the queue group adjusts automatically. - If a worker crashes before deleting its message, the message remains in the mailbox and will be re-delivered when the worker reconnects (store-first semantics). diff --git a/docs/python.md b/docs/python.md index 8d91230..29be72f 100644 --- a/docs/python.md +++ b/docs/python.md @@ -70,9 +70,9 @@ await client.delete(mail_id, msg_id) ## Priority ```python -Priority.HIGH # "high" -Priority.NORMAL # "normal" -Priority.LOW # "low" +Priority.CRITICAL # "critical" — highest priority +Priority.URGENT # "urgent" +Priority.NORMAL # "normal" — default, no suffix ``` ## Public mailbox diff --git a/docs/rust.md b/docs/rust.md index 97852c6..dbfe1a1 100644 --- a/docs/rust.md +++ b/docs/rust.md @@ -69,9 +69,9 @@ client.delete(mail_id: &str, msg_id: &str) -> Result<(), MQ9Error> ## Priority ```rust -Priority::High // "high" -Priority::Normal // "normal" -Priority::Low // "low" +Priority::Critical // "critical" — highest +Priority::Urgent // "urgent" +Priority::Normal // "normal" — default, no suffix ``` ## Public mailbox diff --git a/go/mq9/client.go b/go/mq9/client.go index e9e249c..893f4a1 100644 --- a/go/mq9/client.go +++ b/go/mq9/client.go @@ -22,8 +22,27 @@ const ( subjectCreate = prefix + ".MAILBOX.CREATE" ) -func subjectMsg(mailID, priority string) string { - return fmt.Sprintf("%s.MAILBOX.MSG.%s.%s", prefix, mailID, priority) +func subjectMsgBase(mailID string) string { + return fmt.Sprintf("%s.MAILBOX.MSG.%s", prefix, mailID) +} + +// subjectMsg returns the send subject. Normal uses the bare subject; urgent/critical append a suffix. +func subjectMsg(mailID string, p Priority) string { + if p == Normal { + return subjectMsgBase(mailID) + } + return fmt.Sprintf("%s.%s", subjectMsgBase(mailID), string(p)) +} + +// subjectSub returns the subscribe subject. "*" uses the wildcard suffix. +func subjectSub(mailID, priority string) string { + if priority == "*" { + return subjectMsgBase(mailID) + ".*" + } + if Priority(priority) == Normal { + return subjectMsgBase(mailID) + } + return fmt.Sprintf("%s.%s", subjectMsgBase(mailID), priority) } func subjectList(mailID string) string { return fmt.Sprintf("%s.MAILBOX.LIST.%s", prefix, mailID) @@ -40,9 +59,9 @@ func subjectDelete(mailID, msgID string) string { type Priority string const ( - High Priority = "high" - Normal Priority = "normal" - Low Priority = "low" + Critical Priority = "critical" + Urgent Priority = "urgent" + Normal Priority = "normal" ) // Mailbox represents a created mq9 mailbox. @@ -227,7 +246,7 @@ func (c *MQ9Client) Send(mailID string, payload []byte, priority Priority) error if err := c.ensureConnected(); err != nil { return err } - return c.nc.Publish(subjectMsg(mailID, string(priority)), payload) + return c.nc.Publish(subjectMsg(mailID, priority), payload) } // --------------------------------------------------------------------------- @@ -245,7 +264,7 @@ func (c *MQ9Client) Subscribe(mailID string, callback func(*Message), opts ...Su for _, opt := range opts { opt(o) } - subject := subjectMsg(mailID, o.priority) + subject := subjectSub(mailID, o.priority) handler := func(raw *nats.Msg) { msg := parseIncoming(mailID, raw) @@ -340,12 +359,12 @@ func (c *MQ9Client) request(subject string, payload map[string]any) (map[string] } func parseIncoming(mailID string, raw *nats.Msg) *Message { - parts := strings.Split(raw.Subject, ".") - priorityStr := "normal" - if len(parts) > 0 { - priorityStr = parts[len(parts)-1] - } - priority := toPriority(priorityStr) + // Subject is either the bare form $mq9.AI.MAILBOX.MSG.{mailID} + // or the suffixed form $mq9.AI.MAILBOX.MSG.{mailID}.{urgent|critical} + base := subjectMsgBase(mailID) + suffix := strings.TrimPrefix(raw.Subject, base) + priorityStr := strings.TrimPrefix(suffix, ".") + priority := toPriority(priorityStr) // empty string → Normal // Try JSON envelope var node map[string]any @@ -373,10 +392,10 @@ func parseMessageNode(mailID string, node map[string]any) *Message { func toPriority(s string) Priority { switch s { - case "high": - return High - case "low": - return Low + case "critical": + return Critical + case "urgent": + return Urgent default: return Normal } diff --git a/go/mq9/client_test.go b/go/mq9/client_test.go index 3dbb65e..bdf53be 100644 --- a/go/mq9/client_test.go +++ b/go/mq9/client_test.go @@ -165,17 +165,28 @@ func TestSend_NormalPriority(t *testing.T) { if len(mock.published) != 1 { t.Fatalf("expected 1 publish, got %d", len(mock.published)) } - want := "$mq9.AI.MAILBOX.MSG.m-001.normal" + // normal uses bare subject (no suffix) + want := "$mq9.AI.MAILBOX.MSG.m-001" if mock.published[0].Subject != want { t.Errorf("got subject %q, want %q", mock.published[0].Subject, want) } } -func TestSend_HighPriority(t *testing.T) { +func TestSend_CriticalPriority(t *testing.T) { mock := &mockConn{} c := newClient(mock) - _ = c.Send("m-001", []byte("urgent"), High) - want := "$mq9.AI.MAILBOX.MSG.m-001.high" + _ = c.Send("m-001", []byte("abort"), Critical) + want := "$mq9.AI.MAILBOX.MSG.m-001.critical" + if mock.published[0].Subject != want { + t.Errorf("got %q, want %q", mock.published[0].Subject, want) + } +} + +func TestSend_UrgentPriority(t *testing.T) { + mock := &mockConn{} + c := newClient(mock) + _ = c.Send("m-001", []byte("interrupt"), Urgent) + want := "$mq9.AI.MAILBOX.MSG.m-001.urgent" if mock.published[0].Subject != want { t.Errorf("got %q, want %q", mock.published[0].Subject, want) } @@ -195,7 +206,7 @@ func TestList_Messages(t *testing.T) { "messages": []any{ map[string]any{ "msg_id": "x1", - "priority": "high", + "priority": "critical", "ts": float64(100), }, }, @@ -213,8 +224,8 @@ func TestList_Messages(t *testing.T) { if msgs[0].MsgID != "x1" { t.Errorf("got msg_id %q, want %q", msgs[0].MsgID, "x1") } - if msgs[0].Priority != High { - t.Errorf("got priority %q, want High", msgs[0].Priority) + if msgs[0].Priority != Critical { + t.Errorf("got priority %q, want Critical", msgs[0].Priority) } if msgs[0].Ts != 100 { t.Errorf("got ts %d, want 100", msgs[0].Ts) @@ -255,20 +266,6 @@ func TestDelete(t *testing.T) { } } -// --------------------------------------------------------------------------- -// Tests — Send (low priority) -// --------------------------------------------------------------------------- - -func TestSend_LowPriority(t *testing.T) { - mock := &mockConn{} - c := newClient(mock) - _ = c.Send("m-001", []byte("bg"), Low) - want := "$mq9.AI.MAILBOX.MSG.m-001.low" - if mock.published[0].Subject != want { - t.Errorf("got %q, want %q", mock.published[0].Subject, want) - } -} - // --------------------------------------------------------------------------- // Tests — Subscribe // --------------------------------------------------------------------------- @@ -292,7 +289,26 @@ func TestSubscribe_AllPriorities(t *testing.T) { } } -func TestSubscribe_SinglePriority(t *testing.T) { +func TestSubscribe_SinglePriority_Critical(t *testing.T) { + var capturedSubject string + mock := &mockConn{ + subscribeFn: func(subject string, _ nats.MsgHandler) (*nats.Subscription, error) { + capturedSubject = subject + return &nats.Subscription{}, nil + }, + } + c := newClient(mock) + _, err := c.Subscribe("m-001", func(_ *Message) {}, WithPriority(Critical)) + if err != nil { + t.Fatal(err) + } + want := "$mq9.AI.MAILBOX.MSG.m-001.critical" + if capturedSubject != want { + t.Errorf("got subject %q, want %q", capturedSubject, want) + } +} + +func TestSubscribe_SinglePriority_Normal(t *testing.T) { var capturedSubject string mock := &mockConn{ subscribeFn: func(subject string, _ nats.MsgHandler) (*nats.Subscription, error) { @@ -301,11 +317,12 @@ func TestSubscribe_SinglePriority(t *testing.T) { }, } c := newClient(mock) - _, err := c.Subscribe("m-001", func(_ *Message) {}, WithPriority(High)) + _, err := c.Subscribe("m-001", func(_ *Message) {}, WithPriority(Normal)) if err != nil { t.Fatal(err) } - want := "$mq9.AI.MAILBOX.MSG.m-001.high" + // normal uses bare subject (no suffix) + want := "$mq9.AI.MAILBOX.MSG.m-001" if capturedSubject != want { t.Errorf("got subject %q, want %q", capturedSubject, want) } @@ -347,9 +364,9 @@ func TestSubscribe_CallbackInvoked(t *testing.T) { t.Fatal(err) } - // Simulate server pushing a raw NATS message + // Simulate server pushing a message on the bare subject (normal priority) raw := &nats.Msg{ - Subject: "$mq9.AI.MAILBOX.MSG.m-001.normal", + Subject: "$mq9.AI.MAILBOX.MSG.m-001", Data: []byte(`{"msg_id":"x1","priority":"normal","payload":"aGVsbG8="}`), } capturedHandler(raw) diff --git a/java/pom.xml b/java/pom.xml index 461b4d0..011262a 100644 --- a/java/pom.xml +++ b/java/pom.xml @@ -7,7 +7,7 @@ com.robustmq robustmq - 0.3.5 + 1.0.1 jar RobustMQ Java SDK @@ -164,7 +164,7 @@ central true - published + validated diff --git a/java/src/main/java/com/robustmq/mq9/MQ9Client.java b/java/src/main/java/com/robustmq/mq9/MQ9Client.java index 2253c74..1a6c75a 100644 --- a/java/src/main/java/com/robustmq/mq9/MQ9Client.java +++ b/java/src/main/java/com/robustmq/mq9/MQ9Client.java @@ -108,7 +108,8 @@ private CompletableFuture createInternal(int ttl, bool public CompletableFuture send(String mailId, byte[] payload, Priority priority) { return CompletableFuture.runAsync(() -> { ensureConnected(); - String subject = PREFIX + ".MAILBOX.MSG." + mailId + "." + priority.getValue(); + String base = PREFIX + ".MAILBOX.MSG." + mailId; + String subject = (priority == Priority.NORMAL) ? base : base + "." + priority.getValue(); try { nc.publish(subject, payload); } catch (Exception e) { @@ -123,9 +124,15 @@ public CompletableFuture subscribe(String mailId, MessageHandler han Priority priority, String queueGroup) { return CompletableFuture.supplyAsync(() -> { ensureConnected(); - String sub = (priority != null) - ? PREFIX + ".MAILBOX.MSG." + mailId + "." + priority.getValue() - : PREFIX + ".MAILBOX.MSG." + mailId + ".*"; + String base = PREFIX + ".MAILBOX.MSG." + mailId; + String sub; + if (priority == null) { + sub = base + ".*"; + } else if (priority == Priority.NORMAL) { + sub = base; + } else { + sub = base + "." + priority.getValue(); + } Dispatcher dispatcher = nc.createDispatcher((Message msg) -> { com.robustmq.mq9.Message m = parseIncoming(mailId, msg); @@ -201,11 +208,13 @@ private JsonNode doRequest(String subject, ObjectNode payload) { } private com.robustmq.mq9.Message parseIncoming(String mailId, Message raw) { - // Extract priority from subject last token + // Subject is bare $mq9.AI.MAILBOX.MSG.{mailId} (normal) + // or $mq9.AI.MAILBOX.MSG.{mailId}.{urgent|critical} String subject = raw.getSubject(); - String[] parts = subject.split("\\."); - String priorityStr = parts.length > 0 ? parts[parts.length - 1] : "normal"; - Priority priority = Priority.fromString(priorityStr); + String base = PREFIX + ".MAILBOX.MSG." + mailId; + String suffix = subject.startsWith(base) ? subject.substring(base.length()) : ""; + String priorityStr = suffix.startsWith(".") ? suffix.substring(1) : ""; + Priority priority = priorityStr.isEmpty() ? Priority.NORMAL : Priority.fromString(priorityStr); // Try JSON envelope first try { diff --git a/java/src/main/java/com/robustmq/mq9/Priority.java b/java/src/main/java/com/robustmq/mq9/Priority.java index 834b7fa..03e7fb9 100644 --- a/java/src/main/java/com/robustmq/mq9/Priority.java +++ b/java/src/main/java/com/robustmq/mq9/Priority.java @@ -1,9 +1,9 @@ package com.robustmq.mq9; public enum Priority { - HIGH("high"), - NORMAL("normal"), - LOW("low"); + CRITICAL("critical"), + URGENT("urgent"), + NORMAL("normal"); private final String value; diff --git a/java/src/test/java/com/robustmq/mq9/MQ9ClientTest.java b/java/src/test/java/com/robustmq/mq9/MQ9ClientTest.java index 02aea01..da43549 100644 --- a/java/src/test/java/com/robustmq/mq9/MQ9ClientTest.java +++ b/java/src/test/java/com/robustmq/mq9/MQ9ClientTest.java @@ -1,7 +1,6 @@ package com.robustmq.mq9; import com.fasterxml.jackson.databind.ObjectMapper; -import com.fasterxml.jackson.databind.node.ObjectNode; import io.nats.client.Connection; import io.nats.client.Dispatcher; import org.junit.jupiter.api.BeforeEach; @@ -90,22 +89,20 @@ void createServerError() throws Exception { @Test void sendNormalPriority() throws Exception { client.send("m-001", "hello".getBytes(), Priority.NORMAL).get(); - - verify(mockNc).publish("$mq9.AI.MAILBOX.MSG.m-001.normal", "hello".getBytes()); + // normal uses bare subject (no suffix) + verify(mockNc).publish("$mq9.AI.MAILBOX.MSG.m-001", "hello".getBytes()); } @Test - void sendHighPriority() throws Exception { - client.send("m-001", "urgent".getBytes(), Priority.HIGH).get(); - - verify(mockNc).publish("$mq9.AI.MAILBOX.MSG.m-001.high", "urgent".getBytes()); + void sendCriticalPriority() throws Exception { + client.send("m-001", "abort".getBytes(), Priority.CRITICAL).get(); + verify(mockNc).publish("$mq9.AI.MAILBOX.MSG.m-001.critical", "abort".getBytes()); } @Test - void sendLowPriority() throws Exception { - client.send("m-001", "bg".getBytes(), Priority.LOW).get(); - - verify(mockNc).publish("$mq9.AI.MAILBOX.MSG.m-001.low", "bg".getBytes()); + void sendUrgentPriority() throws Exception { + client.send("m-001", "interrupt".getBytes(), Priority.URGENT).get(); + verify(mockNc).publish("$mq9.AI.MAILBOX.MSG.m-001.urgent", "interrupt".getBytes()); } // ── list ────────────────────────────────────────────────────────────────── @@ -113,7 +110,7 @@ void sendLowPriority() throws Exception { @Test void listMessages() throws Exception { String json = "{\"mail_id\":\"m-001\",\"messages\":[{\"msg_id\":\"x1\"," - + "\"priority\":\"high\",\"ts\":100}]}"; + + "\"priority\":\"critical\",\"ts\":100}]}"; when(mockNc.request(eq("$mq9.AI.MAILBOX.LIST.m-001"), any(byte[].class), any(Duration.class))) .thenReturn(fakeReply(json)); @@ -121,7 +118,7 @@ void listMessages() throws Exception { assertEquals(1, metas.size()); assertEquals("x1", metas.get(0).getMsgId()); - assertEquals(Priority.HIGH, metas.get(0).getPriority()); + assertEquals(Priority.CRITICAL, metas.get(0).getPriority()); assertEquals(100L, metas.get(0).getTs()); } @@ -170,14 +167,26 @@ void subscribeAllPriorities() throws Exception { } @Test - void subscribeSinglePriority() throws Exception { + void subscribeSinglePriority_Critical() throws Exception { + Dispatcher dispatcher = mock(Dispatcher.class); + when(dispatcher.subscribe(anyString())).thenReturn(dispatcher); + when(mockNc.createDispatcher(any())).thenReturn(dispatcher); + + client.subscribe("m-001", msg -> {}, Priority.CRITICAL, "").get(); + + verify(dispatcher).subscribe("$mq9.AI.MAILBOX.MSG.m-001.critical"); + } + + @Test + void subscribeSinglePriority_Normal() throws Exception { Dispatcher dispatcher = mock(Dispatcher.class); when(dispatcher.subscribe(anyString())).thenReturn(dispatcher); when(mockNc.createDispatcher(any())).thenReturn(dispatcher); - client.subscribe("m-001", msg -> {}, Priority.HIGH, "").get(); + client.subscribe("m-001", msg -> {}, Priority.NORMAL, "").get(); - verify(dispatcher).subscribe("$mq9.AI.MAILBOX.MSG.m-001.high"); + // normal uses bare subject (no suffix) + verify(dispatcher).subscribe("$mq9.AI.MAILBOX.MSG.m-001"); } @Test @@ -195,8 +204,6 @@ void subscribeQueueGroup() throws Exception { @Test void closeDrainsConnection() throws Exception { - // nc is already injected; close should attempt drain - // We mock drain to return a CompletableFuture so it doesn't block when(mockNc.drain(any(Duration.class))) .thenReturn(java.util.concurrent.CompletableFuture.completedFuture(true)); diff --git a/javascript/package.json b/javascript/package.json index 1ba77a1..4a6b6ec 100644 --- a/javascript/package.json +++ b/javascript/package.json @@ -1,6 +1,6 @@ { "name": "@robustmq/sdk", - "version": "0.3.5", + "version": "1.0.1", "description": "RobustMQ JavaScript/TypeScript SDK", "main": "dist/index.js", "types": "dist/index.d.ts", diff --git a/javascript/src/mq9/client.test.ts b/javascript/src/mq9/client.test.ts index b4ecd22..b4e46dc 100644 --- a/javascript/src/mq9/client.test.ts +++ b/javascript/src/mq9/client.test.ts @@ -100,24 +100,35 @@ describe("create", () => { // --------------------------------------------------------------------------- describe("send", () => { - test("publishes to correct subject with normal priority", async () => { + test("normal priority uses bare subject (no suffix)", async () => { const t = mockTransport(); const client = clientWith(t); await client.send("m-001", Buffer.from("hello"), "normal"); expect(t.publish).toHaveBeenCalledWith( - "$mq9.AI.MAILBOX.MSG.m-001.normal", + "$mq9.AI.MAILBOX.MSG.m-001", expect.any(Uint8Array), ); }); - test("publishes to high priority subject", async () => { + test("critical priority appends suffix", async () => { const t = mockTransport(); const client = clientWith(t); - await client.send("m-001", Buffer.from("urgent"), "high"); + await client.send("m-001", Buffer.from("abort"), "critical"); expect(t.publish).toHaveBeenCalledWith( - "$mq9.AI.MAILBOX.MSG.m-001.high", + "$mq9.AI.MAILBOX.MSG.m-001.critical", + expect.any(Uint8Array), + ); + }); + + test("urgent priority appends suffix", async () => { + const t = mockTransport(); + const client = clientWith(t); + await client.send("m-001", Buffer.from("interrupt"), "urgent"); + + expect(t.publish).toHaveBeenCalledWith( + "$mq9.AI.MAILBOX.MSG.m-001.urgent", expect.any(Uint8Array), ); }); @@ -141,17 +152,6 @@ describe("send", () => { task: "summarize", }); }); - - test("publishes to low priority subject", async () => { - const t = mockTransport(); - const client = clientWith(t); - await client.send("m-001", Buffer.from("bg"), "low"); - - expect(t.publish).toHaveBeenCalledWith( - "$mq9.AI.MAILBOX.MSG.m-001.low", - expect.any(Uint8Array), - ); - }); }); // --------------------------------------------------------------------------- @@ -196,7 +196,21 @@ describe("subscribe", () => { expect(capturedSubject).toBe("$mq9.AI.MAILBOX.MSG.m-001.*"); }); - test("single priority uses specific subject", async () => { + test("critical priority uses suffixed subject", async () => { + let capturedSubject = ""; + const t = mockTransport({ + subscribe: jest.fn((subject: string) => { + capturedSubject = subject; + return makeSubscription(); + }), + }); + const client = clientWith(t); + await client.subscribe("m-001", async () => {}, { priority: "critical" }); + + expect(capturedSubject).toBe("$mq9.AI.MAILBOX.MSG.m-001.critical"); + }); + + test("normal priority uses bare subject", async () => { let capturedSubject = ""; const t = mockTransport({ subscribe: jest.fn((subject: string) => { @@ -205,9 +219,9 @@ describe("subscribe", () => { }), }); const client = clientWith(t); - await client.subscribe("m-001", async () => {}, { priority: "high" }); + await client.subscribe("m-001", async () => {}, { priority: "normal" }); - expect(capturedSubject).toBe("$mq9.AI.MAILBOX.MSG.m-001.high"); + expect(capturedSubject).toBe("$mq9.AI.MAILBOX.MSG.m-001"); }); test("queue group is forwarded", async () => { @@ -224,10 +238,10 @@ describe("subscribe", () => { expect(capturedOpts.queue).toBe("workers"); }); - test("callback invoked with parsed message", async () => { + test("callback invoked with parsed message from bare subject (normal)", async () => { const payload = Buffer.from("hello").toString("base64"); const raw = { - subject: "$mq9.AI.MAILBOX.MSG.m-001.normal", + subject: "$mq9.AI.MAILBOX.MSG.m-001", data: Buffer.from( JSON.stringify({ msg_id: "x1", priority: "normal", payload }), ), @@ -296,7 +310,7 @@ describe("list", () => { request: jest.fn().mockResolvedValue( jsonReply({ mail_id: "m-001", - messages: [{ msg_id: "x1", priority: "high", ts: 100 }], + messages: [{ msg_id: "x1", priority: "critical", ts: 100 }], }), ), }); @@ -305,7 +319,7 @@ describe("list", () => { expect(msgs).toHaveLength(1); expect(msgs[0].msgId).toBe("x1"); - expect(msgs[0].priority).toBe("high" as Priority); + expect(msgs[0].priority).toBe("critical" as Priority); expect(msgs[0].ts).toBe(100); expect(t.request).toHaveBeenCalledWith( diff --git a/javascript/src/mq9/client.ts b/javascript/src/mq9/client.ts index b2ef5ba..ae81e29 100644 --- a/javascript/src/mq9/client.ts +++ b/javascript/src/mq9/client.ts @@ -8,7 +8,7 @@ import { connect, NatsConnection, Subscription } from "nats"; // Types // --------------------------------------------------------------------------- -export type Priority = "high" | "normal" | "low"; +export type Priority = "critical" | "urgent" | "normal"; export interface Mailbox { mailId: string; @@ -38,7 +38,7 @@ export interface CreateOptions { } export interface SubscribeOptions { - priority?: Priority | "*"; + priority?: Priority | "*"; // "*" = all priorities (default) queueGroup?: string; } @@ -58,8 +58,21 @@ export class MQ9Error extends Error { const PREFIX = "$mq9.AI"; const MAILBOX_CREATE = `${PREFIX}.MAILBOX.CREATE`; +function subjectMsgBase(mailId: string): string { + return `${PREFIX}.MAILBOX.MSG.${mailId}`; +} + +// Send subject: normal uses bare (no suffix); urgent/critical append suffix. function subjectMsg(mailId: string, priority: string): string { - return `${PREFIX}.MAILBOX.MSG.${mailId}.${priority}`; + if (priority === "normal") return subjectMsgBase(mailId); + return `${subjectMsgBase(mailId)}.${priority}`; +} + +// Subscribe subject: "*" uses wildcard; normal uses bare; others append suffix. +function subjectSub(mailId: string, priority: string): string { + if (priority === "*") return `${subjectMsgBase(mailId)}.*`; + if (priority === "normal") return subjectMsgBase(mailId); + return `${subjectMsgBase(mailId)}.${priority}`; } function subjectList(mailId: string): string { return `${PREFIX}.MAILBOX.LIST.${mailId}`; @@ -166,7 +179,7 @@ export class MQ9Client { ): Promise<{ unsubscribe(): void }> { this.ensureConnected(); const priority = opts.priority ?? "*"; - const subject = subjectMsg(mailId, priority); + const subject = subjectSub(mailId, priority); const sub = this._transport!.subscribe(subject, { queue: opts.queueGroup, }); @@ -303,8 +316,9 @@ function parseIncoming( subject: string, data: Uint8Array, ): Mq9Message { - const parts = subject.split("."); - const priorityStr = parts[parts.length - 1] ?? "normal"; + const base = subjectMsgBase(mailId); + const suffix = subject.startsWith(base) ? subject.slice(base.length) : ""; + const priorityStr = suffix.startsWith(".") ? suffix.slice(1) : ""; const priority = toPriority(priorityStr); try { @@ -339,6 +353,6 @@ function parseMessageNode( } function toPriority(s: string): Priority { - if (s === "high" || s === "low") return s; + if (s === "critical" || s === "urgent") return s; return "normal"; } diff --git a/langchain-mq9/README.md b/langchain-mq9/README.md index eb42ed3..11c396d 100644 --- a/langchain-mq9/README.md +++ b/langchain-mq9/README.md @@ -48,7 +48,7 @@ Sends an async message to a mailbox. |-----------|------|---------|-------------| | `mail_id` | str | — | Target mailbox address | | `content` | str | — | Message content | -| `priority` | str | `"normal"` | `"high"`, `"normal"`, or `"low"` | +| `priority` | str | `"normal"` | `"critical"`, `"urgent"`, or `"normal"` | ### `GetMessagesTool` diff --git a/langchain-mq9/examples/agent_example.py b/langchain-mq9/examples/agent_example.py index 92d6163..40af8ba 100644 --- a/langchain-mq9/examples/agent_example.py +++ b/langchain-mq9/examples/agent_example.py @@ -52,7 +52,7 @@ async def main() -> None: agent_b.invoke( { "input": ( - f"Send a high-priority message to mailbox '{mail_id}' " + f"Send a critical-priority message to mailbox '{mail_id}' " "with content: 'Task complete: analysis finished successfully.'" ) } diff --git a/langchain-mq9/langchain_mq9/tools.py b/langchain-mq9/langchain_mq9/tools.py index e2879f3..a7715db 100644 --- a/langchain-mq9/langchain_mq9/tools.py +++ b/langchain-mq9/langchain_mq9/tools.py @@ -31,7 +31,7 @@ class SendMessageInput(BaseModel): content: str = Field(description="The message content to send.") priority: str = Field( default="normal", - description="Message priority: 'high', 'normal', or 'low'.", + description="Message priority: 'critical', 'urgent', or 'normal' (default).", ) @@ -114,8 +114,9 @@ class SendMessageTool(BaseTool): "Use this tool to send an asynchronous message to another Agent or system " "via its mq9 mailbox address (mail_id). " "The recipient does not need to be online — messages are stored until TTL expires. " - "Use priority='high' for urgent tasks that need immediate attention, " - "priority='low' for background work, and priority='normal' (default) for everything else. " + "Use priority='critical' for abort signals or emergencies, " + "priority='urgent' for time-sensitive interrupts, " + "and priority='normal' (default) for routine messages. " "Call this when you want to delegate a task, notify another Agent, " "or pass data to a downstream system." ) diff --git a/langchain-mq9/pyproject.toml b/langchain-mq9/pyproject.toml index 5595ce6..4f7bd57 100644 --- a/langchain-mq9/pyproject.toml +++ b/langchain-mq9/pyproject.toml @@ -4,14 +4,14 @@ build-backend = "hatchling.build" [project] name = "langchain-mq9" -version = "0.3.5" +version = "1.0.1" description = "LangChain tools for the mq9 AI-native async Agent mailbox protocol" readme = "README.md" license = { text = "Apache-2.0" } requires-python = ">=3.10" dependencies = [ "langchain-core>=0.2", - "robustmq>=0.3.5", + "robustmq>=1.0.1", ] [project.optional-dependencies] diff --git a/python/pyproject.toml b/python/pyproject.toml index 8002a65..a972555 100644 --- a/python/pyproject.toml +++ b/python/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "robustmq" -version = "0.3.5" +version = "1.0.1" description = "RobustMQ Python SDK — unified messaging for AI, IoT, and data pipelines" readme = "README.md" license = { text = "Apache-2.0" } diff --git a/python/robustmq/__init__.py b/python/robustmq/__init__.py index 9c47b8d..0b4d00a 100644 --- a/python/robustmq/__init__.py +++ b/python/robustmq/__init__.py @@ -1,3 +1,3 @@ """RobustMQ Python SDK.""" -__version__ = "0.3.5" +__version__ = "1.0.1" diff --git a/python/robustmq/mq9/client.py b/python/robustmq/mq9/client.py index 11b7f06..83b4181 100644 --- a/python/robustmq/mq9/client.py +++ b/python/robustmq/mq9/client.py @@ -21,7 +21,7 @@ # Subject constants _PREFIX = "$mq9.AI" _MAILBOX_CREATE = f"{_PREFIX}.MAILBOX.CREATE" -_MAILBOX_MSG = f"{_PREFIX}.MAILBOX.MSG.{{mail_id}}.{{priority}}" +_MAILBOX_MSG_BASE = f"{_PREFIX}.MAILBOX.MSG.{{mail_id}}" _MAILBOX_LIST = f"{_PREFIX}.MAILBOX.LIST.{{mail_id}}" _MAILBOX_DELETE = f"{_PREFIX}.MAILBOX.DELETE.{{mail_id}}.{{msg_id}}" @@ -29,10 +29,35 @@ _REQUEST_TIMEOUT = 5.0 # seconds +def _msg_subject(mail_id: str, priority: "Priority") -> str: + """Return the NATS subject for sending a message. + + normal (default) uses the bare subject (no suffix). + urgent and critical append the priority as a suffix. + """ + if priority == Priority.NORMAL: + return _MAILBOX_MSG_BASE.format(mail_id=mail_id) + return f"{_MAILBOX_MSG_BASE.format(mail_id=mail_id)}.{priority.value}" + + +def _sub_subject(mail_id: str, priority: str) -> str: + """Return the NATS subject for subscribing. + + "*" → wildcard subject, matches all priorities (normal, urgent, critical). + "normal" → bare subject (no suffix). + "urgent" / "critical" → subject with priority suffix. + """ + if priority == "*": + return f"{_MAILBOX_MSG_BASE.format(mail_id=mail_id)}.*" + if Priority(priority) == Priority.NORMAL: + return _MAILBOX_MSG_BASE.format(mail_id=mail_id) + return f"{_MAILBOX_MSG_BASE.format(mail_id=mail_id)}.{priority}" + + class Priority(str, Enum): - HIGH = "high" + CRITICAL = "critical" + URGENT = "urgent" NORMAL = "normal" - LOW = "low" @dataclass @@ -192,11 +217,11 @@ async def send( Args: mail_id: Target mailbox identifier. payload: Message body. Strings are UTF-8 encoded; dicts are JSON-encoded. - priority: Message priority — high, normal (default), or low. + priority: Message priority — critical, urgent, or normal (default). """ self._ensure_connected() priority = Priority(priority) - subject = _MAILBOX_MSG.format(mail_id=mail_id, priority=priority.value) + subject = _msg_subject(mail_id, priority) data = _encode_payload(payload) await self._nc.publish(subject, data) # type: ignore[union-attr] @@ -222,11 +247,8 @@ async def subscribe( """ self._ensure_connected() - if priority == "*": - subject = _MAILBOX_MSG.format(mail_id=mail_id, priority="*") - else: - p = Priority(priority).value - subject = _MAILBOX_MSG.format(mail_id=mail_id, priority=p) + p = priority if isinstance(priority, str) else priority.value + subject = _sub_subject(mail_id, p) async def _handler(raw: Msg) -> None: try: @@ -339,11 +361,15 @@ def _parse_incoming(mail_id: str, raw: Msg) -> Message: The server may send a JSON envelope or raw bytes depending on version. We try JSON first and fall back to treating the full body as payload. """ - # Extract priority from subject: last token - parts = raw.subject.split(".") - priority_str = parts[-1] if parts else "normal" + # Extract priority from subject. + # Subject is $mq9.AI.MAILBOX.MSG.{mail_id} (normal, no suffix) + # or $mq9.AI.MAILBOX.MSG.{mail_id}.{urgent|critical} + # We strip the fixed prefix to find whether a priority suffix is present. + _msg_prefix = f"$mq9.AI.MAILBOX.MSG.{mail_id}" + suffix = raw.subject[len(_msg_prefix):] # "" or ".urgent" or ".critical" + priority_str = suffix.lstrip(".") try: - priority = Priority(priority_str) + priority = Priority(priority_str) if priority_str else Priority.NORMAL except ValueError: priority = Priority.NORMAL diff --git a/python/tests/test_client.py b/python/tests/test_client.py index 1602575..45eb58e 100644 --- a/python/tests/test_client.py +++ b/python/tests/test_client.py @@ -160,7 +160,8 @@ async def test_send_bytes_normal_priority(): await client.send("m-001", b"hello", priority=Priority.NORMAL) - nc.publish.assert_awaited_once_with("$mq9.AI.MAILBOX.MSG.m-001.normal", b"hello") + # normal uses bare subject (no suffix) + nc.publish.assert_awaited_once_with("$mq9.AI.MAILBOX.MSG.m-001", b"hello") @pytest.mark.asyncio @@ -179,21 +180,21 @@ async def test_send_dict_payload(): client = _make_client() nc = await _inject_nc(client) - await client.send("m-001", {"task": "summarize"}, priority="high") + await client.send("m-001", {"task": "summarize"}, priority="critical") subject, data = nc.publish.call_args.args - assert subject == "$mq9.AI.MAILBOX.MSG.m-001.high" + assert subject == "$mq9.AI.MAILBOX.MSG.m-001.critical" assert json.loads(data) == {"task": "summarize"} @pytest.mark.asyncio -async def test_send_low_priority(): +async def test_send_urgent_priority(): client = _make_client() nc = await _inject_nc(client) - await client.send("m-001", b"bg", priority=Priority.LOW) + await client.send("m-001", b"interrupt", priority=Priority.URGENT) - nc.publish.assert_awaited_once_with("$mq9.AI.MAILBOX.MSG.m-001.low", b"bg") + nc.publish.assert_awaited_once_with("$mq9.AI.MAILBOX.MSG.m-001.urgent", b"interrupt") @pytest.mark.asyncio @@ -202,7 +203,7 @@ async def test_send_invalid_priority(): await _inject_nc(client) with pytest.raises(ValueError): - await client.send("m-001", b"x", priority="urgent") + await client.send("m-001", b"x", priority="extreme") # --------------------------------------------------------------------------- @@ -226,15 +227,28 @@ async def test_subscribe_all_priorities(): @pytest.mark.asyncio -async def test_subscribe_single_priority(): +async def test_subscribe_single_priority_critical(): client = _make_client() nc = await _inject_nc(client) nc.subscribe = AsyncMock(return_value=MagicMock()) - await client.subscribe("m-001", AsyncMock(), priority="high") + await client.subscribe("m-001", AsyncMock(), priority="critical") subject = nc.subscribe.call_args.args[0] - assert subject == "$mq9.AI.MAILBOX.MSG.m-001.high" + assert subject == "$mq9.AI.MAILBOX.MSG.m-001.critical" + + +@pytest.mark.asyncio +async def test_subscribe_single_priority_normal(): + client = _make_client() + nc = await _inject_nc(client) + nc.subscribe = AsyncMock(return_value=MagicMock()) + + await client.subscribe("m-001", AsyncMock(), priority="normal") + + subject = nc.subscribe.call_args.args[0] + # normal uses bare subject (no suffix) + assert subject == "$mq9.AI.MAILBOX.MSG.m-001" @pytest.mark.asyncio @@ -272,9 +286,9 @@ async def handler(msg: Message) -> None: await client.subscribe("m-001", handler, priority="normal") assert captured_cb is not None - # Simulate server pushing a raw message + # Simulate server pushing a raw message on the bare subject (normal) raw = MagicMock() - raw.subject = "$mq9.AI.MAILBOX.MSG.m-001.normal" + raw.subject = "$mq9.AI.MAILBOX.MSG.m-001" raw.data = b'{"msg_id":"x1","priority":"normal","payload":"aGVsbG8="}' await captured_cb(raw) @@ -298,7 +312,7 @@ async def test_list_messages(): server_resp = { "mail_id": "m-001", "messages": [ - {"msg_id": "msg-1", "priority": "high", "ts": 1000}, + {"msg_id": "msg-1", "priority": "critical", "ts": 1000}, ], } nc.request = AsyncMock(return_value=_make_reply(server_resp)) @@ -307,7 +321,7 @@ async def test_list_messages(): assert len(metas) == 1 assert metas[0].msg_id == "msg-1" - assert metas[0].priority == Priority.HIGH + assert metas[0].priority == Priority.CRITICAL assert metas[0].ts == 1000 subject = nc.request.call_args.args[0] @@ -401,34 +415,43 @@ def test_encode_dict(): # --------------------------------------------------------------------------- -def test_parse_incoming_with_envelope(): +def test_parse_incoming_with_envelope_critical(): raw = MagicMock() - raw.subject = "$mq9.AI.MAILBOX.MSG.m-001.high" + raw.subject = "$mq9.AI.MAILBOX.MSG.m-001.critical" raw.data = json.dumps( { "msg_id": "x9", - "priority": "high", + "priority": "critical", "payload": base64.b64encode(b"data").decode(), } ).encode() msg = _parse_incoming("m-001", raw) assert msg.msg_id == "x9" - assert msg.priority == Priority.HIGH + assert msg.priority == Priority.CRITICAL assert msg.payload == b"data" -def test_parse_incoming_raw_bytes(): +def test_parse_incoming_normal_bare_subject(): raw = MagicMock() - raw.subject = "$mq9.AI.MAILBOX.MSG.m-001.low" + raw.subject = "$mq9.AI.MAILBOX.MSG.m-001" raw.data = b"plain bytes" msg = _parse_incoming("m-001", raw) assert msg.payload == b"plain bytes" - assert msg.priority == Priority.LOW + assert msg.priority == Priority.NORMAL assert msg.msg_id == "" +def test_parse_incoming_urgent(): + raw = MagicMock() + raw.subject = "$mq9.AI.MAILBOX.MSG.m-001.urgent" + raw.data = b"interrupt" + + msg = _parse_incoming("m-001", raw) + assert msg.priority == Priority.URGENT + + def test_parse_incoming_unknown_priority(): raw = MagicMock() raw.subject = "$mq9.AI.MAILBOX.MSG.m-001.unknown" diff --git a/rust/Cargo.lock b/rust/Cargo.lock index d04e6d3..6eda26a 100644 --- a/rust/Cargo.lock +++ b/rust/Cargo.lock @@ -754,7 +754,7 @@ dependencies = [ [[package]] name = "robustmq" -version = "0.3.5" +version = "1.0.1" dependencies = [ "async-nats", "bytes", diff --git a/rust/Cargo.toml b/rust/Cargo.toml index 766e96c..a55b4cb 100644 --- a/rust/Cargo.toml +++ b/rust/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "robustmq" -version = "0.3.5" +version = "1.0.1" edition = "2021" description = "RobustMQ Rust SDK — unified messaging for AI, IoT, and data pipelines" license = "Apache-2.0" diff --git a/rust/src/mq9/mod.rs b/rust/src/mq9/mod.rs index 5a816e9..d948e6a 100644 --- a/rust/src/mq9/mod.rs +++ b/rust/src/mq9/mod.rs @@ -33,8 +33,26 @@ use serde_json::Value; const PREFIX: &str = "$mq9.AI"; const MAILBOX_CREATE: &str = "$mq9.AI.MAILBOX.CREATE"; -fn subject_msg(mail_id: &str, priority: &str) -> String { - format!("{PREFIX}.MAILBOX.MSG.{mail_id}.{priority}") +fn subject_msg_base(mail_id: &str) -> String { + format!("{PREFIX}.MAILBOX.MSG.{mail_id}") +} + +/// Send subject: normal uses bare (no suffix); urgent/critical append suffix. +fn subject_msg(mail_id: &str, priority: Priority) -> String { + if priority == Priority::Normal { + subject_msg_base(mail_id) + } else { + format!("{}.{}", subject_msg_base(mail_id), priority.as_str()) + } +} + +/// Subscribe subject: "*" uses wildcard; normal uses bare; others append suffix. +fn subject_sub(mail_id: &str, priority: &str) -> String { + match priority { + "*" => format!("{}.*", subject_msg_base(mail_id)), + "normal" => subject_msg_base(mail_id), + p => format!("{}.{}", subject_msg_base(mail_id), p), + } } fn subject_list(mail_id: &str) -> String { @@ -52,24 +70,24 @@ fn subject_delete(mail_id: &str, msg_id: &str) -> String { /// Message priority level. #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] pub enum Priority { - High, + Critical, + Urgent, Normal, - Low, } impl Priority { pub fn as_str(&self) -> &'static str { match self { - Priority::High => "high", + Priority::Critical => "critical", + Priority::Urgent => "urgent", Priority::Normal => "normal", - Priority::Low => "low", } } pub fn from_str_lossy(s: &str) -> Priority { match s { - "high" => Priority::High, - "low" => Priority::Low, + "critical" => Priority::Critical, + "urgent" => Priority::Urgent, _ => Priority::Normal, } } @@ -347,7 +365,7 @@ impl MQ9Client { payload: &[u8], priority: Priority, ) -> Result<(), MQ9Error> { - let subject = subject_msg(mail_id, priority.as_str()); + let subject = subject_msg(mail_id, priority); self.transport .publish(subject, Bytes::copy_from_slice(payload)) .await @@ -373,7 +391,7 @@ impl MQ9Client { let priority_str = priority .map(|p| p.as_str().to_string()) .unwrap_or_else(|| "*".to_string()); - let subject = subject_msg(mail_id, &priority_str); + let subject = subject_sub(mail_id, &priority_str); let queue = if queue_group.is_empty() { None } else { @@ -447,14 +465,13 @@ impl MQ9Client { /// Parse a message arriving on a subscription subject. /// -/// Subject format: `$mq9.AI.MAILBOX.MSG.{mail_id}.{priority}` +/// Subject is bare `$mq9.AI.MAILBOX.MSG.{mail_id}` (normal) +/// or `$mq9.AI.MAILBOX.MSG.{mail_id}.{urgent|critical}`. fn parse_incoming(mail_id: &str, raw: &RawMsg) -> Message { - let priority = raw - .subject - .rsplit('.') - .next() - .map(Priority::from_str_lossy) - .unwrap_or(Priority::Normal); + let base = subject_msg_base(mail_id); + let suffix = raw.subject.strip_prefix(&base).unwrap_or(""); + let priority_str = suffix.strip_prefix('.').unwrap_or(""); + let priority = Priority::from_str_lossy(priority_str); // Try JSON envelope first if let Ok(envelope) = serde_json::from_slice::(&raw.payload) { @@ -1099,15 +1116,15 @@ mod tests { #[test] fn test_priority_as_str() { - assert_eq!(Priority::High.as_str(), "high"); + assert_eq!(Priority::Critical.as_str(), "critical"); + assert_eq!(Priority::Urgent.as_str(), "urgent"); assert_eq!(Priority::Normal.as_str(), "normal"); - assert_eq!(Priority::Low.as_str(), "low"); } #[test] fn test_priority_from_str_lossy() { - assert_eq!(Priority::from_str_lossy("high"), Priority::High); - assert_eq!(Priority::from_str_lossy("low"), Priority::Low); + assert_eq!(Priority::from_str_lossy("critical"), Priority::Critical); + assert_eq!(Priority::from_str_lossy("urgent"), Priority::Urgent); assert_eq!(Priority::from_str_lossy("normal"), Priority::Normal); assert_eq!(Priority::from_str_lossy("unknown"), Priority::Normal); assert_eq!(Priority::from_str_lossy("*"), Priority::Normal); @@ -1115,9 +1132,9 @@ mod tests { #[test] fn test_priority_display() { - assert_eq!(format!("{}", Priority::High), "high"); + assert_eq!(format!("{}", Priority::Critical), "critical"); + assert_eq!(format!("{}", Priority::Urgent), "urgent"); assert_eq!(format!("{}", Priority::Normal), "normal"); - assert_eq!(format!("{}", Priority::Low), "low"); } // ------------------------------------------------------------------ @@ -1153,15 +1170,31 @@ mod tests { #[test] fn test_subject_msg_normal() { assert_eq!( - subject_msg("m-001", "normal"), - "$mq9.AI.MAILBOX.MSG.m-001.normal" + subject_msg("m-001", Priority::Normal), + "$mq9.AI.MAILBOX.MSG.m-001" + ); + } + + #[test] + fn test_subject_msg_urgent() { + assert_eq!( + subject_msg("m-001", Priority::Urgent), + "$mq9.AI.MAILBOX.MSG.m-001.urgent" + ); + } + + #[test] + fn test_subject_msg_critical() { + assert_eq!( + subject_msg("m-001", Priority::Critical), + "$mq9.AI.MAILBOX.MSG.m-001.critical" ); } #[test] - fn test_subject_msg_wildcard() { + fn test_subject_sub_wildcard() { assert_eq!( - subject_msg("task.queue", "*"), + subject_sub("task.queue", "*"), "$mq9.AI.MAILBOX.MSG.task.queue.*" ); }