Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 6 additions & 5 deletions CLAUDE.md
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down Expand Up @@ -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
Expand Down
25 changes: 13 additions & 12 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -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. |

Expand All @@ -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}` |
Expand All @@ -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

Expand Down Expand Up @@ -99,7 +100,7 @@ await client.send(mailbox.mailId, "hello", "normal");
<dependency>
<groupId>com.robustmq</groupId>
<artifactId>robustmq</artifactId>
<version>0.3.5</version>
<version>1.0.1</version>
</dependency>
```
```java
Expand All @@ -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**
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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)
```

---
Expand Down
2 changes: 1 addition & 1 deletion VERSION
Original file line number Diff line number Diff line change
@@ -1 +1 @@
0.3.5
1.0.1
47 changes: 34 additions & 13 deletions csharp/RobustMQ.Mq9.Tests/MQ9ClientTests.cs
Original file line number Diff line number Diff line change
Expand Up @@ -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<INatsConnection>();
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<byte[]>());
}

[Fact]
public async Task Send_HighPriority_CorrectSubject()
public async Task Send_CriticalPriority_UsesSuffix()
{
var mock = Substitute.For<INatsConnection>();
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<byte[]>());
}

[Fact]
public async Task Send_LowPriority_CorrectSubject()
public async Task Send_UrgentPriority_UsesSuffix()
{
var mock = Substitute.For<INatsConnection>();
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<byte[]>());
}

Expand All @@ -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<INatsConnection>();
var capturedSubject = "";
Expand All @@ -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<INatsConnection>();
var capturedSubject = "";
mock.SubscribeCoreAsync<byte[]>(Arg.Any<string>(), cancellationToken: Arg.Any<CancellationToken>())
.Returns(callInfo =>
{
capturedSubject = callInfo.ArgAt<string>(0);
return ValueTask.FromResult(Substitute.For<INatsSub<byte[]>>());
});
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]
Expand Down Expand Up @@ -210,7 +231,7 @@ await Assert.ThrowsAsync<InvalidOperationException>(
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<INatsConnection>();
mock.RequestAsync<byte[], byte[]>(
Expand All @@ -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);
}

Expand Down
16 changes: 11 additions & 5 deletions csharp/RobustMQ.Mq9/MQ9Client.cs
Original file line number Diff line number Diff line change
Expand Up @@ -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);
}

Expand All @@ -96,7 +98,9 @@ public async Task<IAsyncDisposable> 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)
Expand Down Expand Up @@ -182,9 +186,11 @@ private async Task<JsonNode> 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
Expand Down
15 changes: 7 additions & 8 deletions csharp/RobustMQ.Mq9/Models.cs
Original file line number Diff line number Diff line change
Expand Up @@ -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
};
}
Expand Down
2 changes: 1 addition & 1 deletion csharp/RobustMQ.Mq9/RobustMQ.Mq9.csproj
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,7 @@
<PackageId>RobustMQ</PackageId>
<RootNamespace>RobustMQ.Mq9</RootNamespace>
<AssemblyName>RobustMQ</AssemblyName>
<Version>0.3.5</Version>
<Version>1.0.1</Version>
<Description>RobustMQ SDK — unified messaging for AI, IoT, and data pipelines</Description>
<PackageLicenseExpression>Apache-2.0</PackageLicenseExpression>
<PackageProjectUrl>https://github.com/robustmq/robustmq-sdk</PackageProjectUrl>
Expand Down
6 changes: 3 additions & 3 deletions demo/demo-csharp/Program.cs
Original file line number Diff line number Diff line change
Expand Up @@ -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 =>
Expand Down
2 changes: 1 addition & 1 deletion demo/demo-csharp/demo-csharp.csproj
Original file line number Diff line number Diff line change
Expand Up @@ -8,7 +8,7 @@
</PropertyGroup>

<ItemGroup>
<PackageReference Include="RobustMQ.Mq9" Version="0.3.5" />
<PackageReference Include="RobustMQ.Mq9" Version="1.0.1" />
</ItemGroup>

</Project>
6 changes: 3 additions & 3 deletions demo/demo-go/main.go
Original file line number Diff line number Diff line change
Expand Up @@ -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) {
Expand Down
4 changes: 2 additions & 2 deletions demo/demo-java/pom.xml
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,7 @@

<groupId>io.robustmq.demo</groupId>
<artifactId>demo-java</artifactId>
<version>0.3.5</version>
<version>1.0.1</version>
<packaging>jar</packaging>

<name>RobustMQ mq9 Java Demo</name>
Expand All @@ -22,7 +22,7 @@
<dependency>
<groupId>com.robustmq</groupId>
<artifactId>robustmq</artifactId>
<version>0.3.5</version>
<version>1.0.1</version>
</dependency>
</dependencies>

Expand Down
6 changes: 3 additions & 3 deletions demo/demo-java/src/main/java/com/robustmq/demo/Demo.java
Original file line number Diff line number Diff line change
Expand Up @@ -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 ->
Expand Down
6 changes: 3 additions & 3 deletions demo/demo-javascript/demo.js
Original file line number Diff line number Diff line change
Expand Up @@ -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) => {
Expand Down
2 changes: 1 addition & 1 deletion demo/demo-javascript/package.json
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,6 @@
"start": "node demo.js"
},
"dependencies": {
"@robustmq/sdk": "0.3.5"
"@robustmq/sdk": "1.0.1"
}
}
Loading
Loading