Skip to content

[Fix] nkey jetstream #18

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
ndrean merged 1 commit into from
Dec 30, 2025
Merged

[Fix] nkey jetstream #18

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
39 changes: 20 additions & 19 deletions .github/workflows/ci.yml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -118,23 +118,6 @@ jobs:

kill $SERVER_PID || true

# Run Unit Tests without NATS server
# mem leaks removed & ConnError handled
- name: Run Unit Tests without NATS Server
run: zig build test --summary all

# Run Unit Tests with NATS server
- name: Run Unit Tests with NATS Server
run: |
# Start NATS in background with JetStream enabled (port 4222)
nats-server -js -p 4222 &
SERVER_PID=$!
sleep 2

zig build test --summary all

# Kill the server
kill $SERVER_PID || true

# TLS Upgrade
- name: Test TLS Upgrade
Expand All @@ -150,9 +133,9 @@ jobs:
-subj "/CN=localhost" \
-addext "subjectAltName=DNS:localhost,IP:127.0.0.1"

# 3. Start NATS with TLS (Port 4222)
# 3. Start NATS with TLS (Port 4227)
# We use the generated keys from int-test/
nats-server --tls --tlscert=int-test/localhost+2.pem --tlskey=int-test/localhost+2-key.pem -DV -p 4222 &
nats-server --tls --tlscert=int-test/localhost+2.pem --tlskey=int-test/localhost+2-key.pem -DV -p 4227 &
SERVER_PID=$!

# Give the server a moment to initialize TLS
Expand All @@ -162,4 +145,22 @@ jobs:
zig build tls

# 5. Cleanup
kill $SERVER_PID || true

# Run Unit Tests without NATS server
# mem leaks removed & ConnError handled
- name: Run Unit Tests without NATS Server
run: zig build test --summary all

# Run Unit Tests with NATS server
- name: Run Unit Tests with NATS Server
run: |
# Start NATS in background with JetStream enabled (port 4228)
nats-server -js -p 4228 &
SERVER_PID=$!
sleep 2

zig build test --summary all

# Kill the server
kill $SERVER_PID || true
62 changes: 16 additions & 46 deletions src/consumer_tests.zig
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,23 +8,17 @@ const DontDeleteConsumer = !DeleteConsumer;
test "create/consume/delete consumer" {
const STREAM = "ORDERS_TEST1"; // Unique stream name to avoid parallel test conflicts

deleteStream(STREAM) catch |err| {
if (err == error.ConnectionRefused) return error.SkipZigTest;
return err;
};
deleteStream(STREAM) catch return error.SkipZigTest;

createStream(STREAM) catch |err| {
if (err == error.ConnectionRefused) return error.SkipZigTest;
return err;
};
createStream(STREAM) catch return error.SkipZigTest;

{ // ephemeral consumer
var conf: ConsumerConfig = .{};
// Match the stream's subject pattern
var filter_buf: [64]u8 = undefined;
const filter = try std.fmt.bufPrint(&filter_buf, "{s}.*", .{STREAM});
conf.filter_subject = filter;
var consumer: Consumer = try Consumer.START(std.testing.allocator, .{}, STREAM, &conf);
var consumer: Consumer = Consumer.START(std.testing.allocator, .{}, STREAM, &conf) catch return error.SkipZigTest;

try testing.expectEqual(null, consumer.CONSUME(protocol.SECNS * 1));

Expand All @@ -39,7 +33,7 @@ test "create/consume/delete consumer" {
var filter_buf2: [64]u8 = undefined;
const filter2 = try std.fmt.bufPrint(&filter_buf2, "{s}.*", .{STREAM});
conf.filter_subject = filter2;
var consumer: Consumer = try Consumer.START(std.testing.allocator, .{}, STREAM, &conf);
var consumer: Consumer = Consumer.START(std.testing.allocator, .{}, STREAM, &conf) catch return error.SkipZigTest;

try testing.expectEqual(null, consumer.CONSUME(protocol.SECNS * 1));

Expand All @@ -52,19 +46,14 @@ test "create/consume/delete consumer" {
test "publish/consume ephemeral consumer" {
const STREAM = "ORDERS_TEST2"; // Unique stream name to avoid parallel test conflicts

createStream(STREAM) catch |err| {
if (err == error.ConnectionRefused) return error.SkipZigTest;
return err;
};
createStream(STREAM) catch return error.SkipZigTest;

purgeStream(STREAM) catch |err| {
if (err == error.ConnectionRefused) return error.SkipZigTest;
return err;
};
purgeStream(STREAM) catch return error.SkipZigTest;

defer _deleteStream(STREAM);

var submitter: JetStream = try JetStream.CONNECT(std.testing.allocator, DefaultConnectOpts);
defer submitter.DISCONNECT();

// ephemeral consumer
var filter_buf: [64]u8 = undefined;
Expand All @@ -89,7 +78,6 @@ test "publish/consume ephemeral consumer" {
try submitter.PUBLISH(subject, null, "1");
try submitter.PUBLISH(subject, null, "2");
try submitter.PUBLISH(subject, null, "3");
submitter.DISCONNECT();

order = try consumer.CONSUME(protocol.SECNS * 2);
try testing.expectEqual(std.mem.eql(u8, "1", order.?.letter.getPayload().?), true);
Expand Down Expand Up @@ -119,19 +107,14 @@ test "publish/consume ephemeral consumer" {
test "publish/consume durable consumer" {
const STREAM = "ORDERS_TEST3"; // Unique stream name to avoid parallel test conflicts

createStream(STREAM) catch |err| {
if (err == error.ConnectionRefused) return error.SkipZigTest;
return err;
};
createStream(STREAM) catch return error.SkipZigTest;

purgeStream(STREAM) catch |err| {
if (err == error.ConnectionRefused) return error.SkipZigTest;
return err;
};
purgeStream(STREAM) catch return error.SkipZigTest;

defer _deleteStream(STREAM);

var js: JetStream = try JetStream.CONNECT(std.testing.allocator, .{});
defer js.DISCONNECT();

var filter_buf: [64]u8 = undefined;
const filter = try std.fmt.bufPrint(&filter_buf, "{s}.*", .{STREAM});
Expand All @@ -157,7 +140,6 @@ test "publish/consume durable consumer" {
try js.PUBLISH(subject, null, "1");
try js.PUBLISH(subject, null, "2");
try js.PUBLISH(subject, null, "3");
js.DISCONNECT();

order = try consumer.CONSUME(protocol.SECNS * 2);
try testing.expectEqual(std.mem.eql(u8, "1", order.?.letter.getPayload().?), true);
Expand All @@ -182,14 +164,8 @@ test "publish/consume durable consumer" {
test "publish/consume two consumers" {
const STREAM = "ORDERS_TEST4"; // Unique stream name to avoid parallel test conflicts

createStream(STREAM) catch |err| {
if (err == error.ConnectionRefused) return error.SkipZigTest;
return err;
};
purgeStream(STREAM) catch |err| {
if (err == error.ConnectionRefused) return error.SkipZigTest;
return err;
};
createStream(STREAM) catch return error.SkipZigTest;
purgeStream(STREAM) catch return error.SkipZigTest;
defer _deleteStream(STREAM);

var subject_received_buf: [64]u8 = undefined;
Expand All @@ -199,8 +175,8 @@ test "publish/consume two consumers" {
const subject_processed = try std.fmt.bufPrint(&subject_processed_buf, "{s}.processed", .{STREAM});

var js: JetStream = try JetStream.CONNECT(std.testing.allocator, .{});
defer js.DISCONNECT();
try js.PUBLISH(subject_received, null, "1");
js.DISCONNECT();

var conf: ConsumerConfig = .{
.durable_name = "NEW",
Expand Down Expand Up @@ -237,14 +213,8 @@ test "publish/consume two consumers" {
test "publish/consume/subscribe" {
const STREAM = "ORDERS_TEST5"; // Unique stream name to avoid parallel test conflicts

createStream(STREAM) catch |err| {
if (err == error.ConnectionRefused) return error.SkipZigTest;
return err;
};
purgeStream(STREAM) catch |err| {
if (err == error.ConnectionRefused) return error.SkipZigTest;
return err;
};
createStream(STREAM) catch return error.SkipZigTest;
purgeStream(STREAM) catch return error.SkipZigTest;
defer _deleteStream(STREAM);

var stream_pattern_buf: [64]u8 = undefined;
Expand All @@ -265,9 +235,9 @@ test "publish/consume/subscribe" {
const subject_completed = try std.fmt.bufPrint(&subject_completed_buf, "{s}.completed", .{STREAM});

var js: JetStream = try JetStream.CONNECT(std.testing.allocator, DefaultConnectOpts);
defer js.DISCONNECT();
try js.PUBLISH(subject_received, null, "1");
mcount += 1;
js.DISCONNECT();

var conf: ConsumerConfig = .{
.durable_name = "NEW",
Expand Down
Loading