Compare commits

...

3 Commits

Author SHA1 Message Date
a53b32204d
tests: add subscription method coverage
This resulted in some minor binding fixes. In theory, all the main
wrapped API endpoints are now covered.
2023-09-03 16:28:00 -07:00
373616f234
tests.message: improve API coverage 2023-09-03 16:14:37 -07:00
6dc50530f1
message: a bit of cleanup around header iterators 2023-09-03 16:13:58 -07:00
7 changed files with 409 additions and 155 deletions

View File

@ -11,24 +11,25 @@ pub fn main() !void {
const message = try nats.Message.create("subject", null, "message");
defer message.destroy();
try message.setHeaderValue("My-Key1", "value1");
try message.setHeaderValue("My-Key2", "value2");
try message.addHeaderValue("My-Key1", "value3");
try message.setHeaderValue("My-Key3", "value4");
try message.setHeaderValue("foo", "foo-value");
try message.setHeaderValue("bar", "bar-value");
try message.addHeaderValue("foo", "bar-value");
try message.setHeaderValue("baz", "baz-value");
try message.addHeaderValue("qux", "qux-value");
try message.deleteHeader("My-Key3");
try message.deleteHeader("baz");
{
var iter = try message.headerIterator();
var iter = try message.getHeaderIterator();
defer iter.destroy();
while (iter.next()) |resolv| {
var val_iter = try resolv.getValueIterator();
while (iter.next()) |header| {
var val_iter = try header.valueIterator();
defer val_iter.destroy();
std.debug.print("key '{s}' got: ", .{resolv.key});
std.debug.print("key '{s}' got: ", .{header.key});
while (val_iter.next()) |value| {
std.debug.print("'{s}', ", .{value});
std.debug.print("'{s}'{s}", .{ value, if (val_iter.peek()) |_| ", " else "" });
}
std.debug.print("\n", .{});
}
@ -42,12 +43,12 @@ pub fn main() !void {
defer received.destroy();
{
var iter = try received.getHeaderValueIterator("My-Key1");
var iter = try received.getHeaderValueIterator("foo");
defer iter.destroy();
std.debug.print("For key 'My-Key1' got: ", .{});
std.debug.print("For key 'foo' got: ", .{});
while (iter.next()) |value| {
std.debug.print("'{s}', ", .{value});
std.debug.print("'{s}'{s}", .{ value, if (iter.peek()) |_| ", " else "" });
}
std.debug.print("\n", .{});
}

View File

@ -423,7 +423,7 @@ pub const Connection = opaque {
) Error!*Subscription {
var sub: *Subscription = undefined;
const status = Status.fromInt(nats_c.natsConnection_QueueSubscribe(
const status = Status.fromInt(nats_c.natsConnection_QueueSubscribeTimeout(
@ptrCast(&sub),
@ptrCast(self),
subject.ptr,
@ -443,7 +443,7 @@ pub const Connection = opaque {
) Error!*Subscription {
var sub: *Subscription = undefined;
const status = Status.fromInt(nats_c.natsConnection_SubscribeSync(
const status = Status.fromInt(nats_c.natsConnection_QueueSubscribeSync(
@ptrCast(&sub),
@ptrCast(self),
subject.ptr,

View File

@ -76,102 +76,11 @@ pub const Message = opaque {
return status.toError() orelse std.mem.sliceTo(value.?, 0);
}
pub fn getAllHeaderValues(self: *Message, key: [:0]const u8) Error![][*:0]const u8 {
var values: [*c][*c]const u8 = undefined;
var count: c_int = 0;
const status = Status.fromInt(nats_c.natsMsgHeader_Values(@ptrCast(self), key.ptr, &values, &count));
// the user must use std.mem.spanTo on each item they want to read to get a
// slice, since we can't do that automatically without having to allocate.
return status.toError() orelse blk: {
const coerced: [*][*:0]const u8 = @ptrFromInt(@intFromPtr(values));
break :blk coerced[0..@intCast(count)];
};
}
pub fn getHeaderValueIterator(self: *Message, key: [:0]const u8) Error!HeaderValueIterator {
return .{ .values = try self.getAllHeaderValues(key) };
}
pub fn getAllHeaderKeys(self: *Message) Error![][*:0]const u8 {
var keys: [*c][*c]const u8 = undefined;
var count: c_int = 0;
const status = Status.fromInt(nats_c.natsMsgHeader_Keys(@ptrCast(self), &keys, &count));
// TODO: manually assert no keys are NULL?
// the user must use std.mem.spanTo on each item they want to read to get a
// slice, since we can't do that automatically without having to allocate.
// the returned slice
return status.toError() orelse blk: {
const coerced: [*][*:0]const u8 = @ptrFromInt(@intFromPtr(keys));
break :blk coerced[0..@intCast(count)];
};
}
pub const HeaderValueIterator = struct {
values: [][*:0]const u8,
index: usize = 0,
pub fn destroy(self: HeaderValueIterator) void {
std.heap.raw_c_allocator.free(self.values);
}
pub fn next(self: *HeaderValueIterator) ?[:0]const u8 {
if (self.index >= self.values.len) return null;
defer self.index += 1;
return std.mem.sliceTo(self.values[self.index], 0);
}
};
pub const HeaderIterator = struct {
message: *Message,
keys: [][*:0]const u8,
index: usize = 0,
pub const ValueResolver = struct {
message: *Message,
key: [:0]const u8,
pub fn getValue(self: ValueResolver) Error![:0]const u8 {
// TODO: if we didn't care about the lifecycle of self.message, we
// could do catch unreachable here and make this error-free
return try self.message.getHeaderValue(self.key);
}
pub fn getValueIterator(self: ValueResolver) Error!HeaderValueIterator {
return .{
.values = try self.message.getAllHeaderValues(self.key),
};
}
};
pub fn destroy(self: *HeaderIterator) void {
std.heap.raw_c_allocator.free(self.keys);
}
pub fn next(self: *HeaderIterator) ?ValueResolver {
if (self.index >= self.keys.len) return null;
defer self.index += 1;
const sliced = std.mem.sliceTo(self.keys[self.index], 0);
return .{
.message = self.message,
.key = sliced,
};
}
pub fn nextKey(self: *HeaderIterator) ?[:0]const u8 {
if (self.index >= self.keys.len) return null;
defer self.index += 1;
return std.mem.sliceTo(self.keys[self.index], 0);
}
};
pub fn headerIterator(self: *Message) Error!HeaderIterator {
pub fn getHeaderIterator(self: *Message) Error!HeaderIterator {
return .{
.message = self,
.keys = try self.getAllHeaderKeys(),
@ -186,6 +95,112 @@ pub const Message = opaque {
pub fn isNoResponders(self: *Message) bool {
return nats_c.natsMsg_IsNoResponders(@ptrCast(self));
}
// prefer using message.getHeaderValueIterator
pub fn getAllHeaderValues(self: *Message, key: [:0]const u8) Error![][*:0]const u8 {
var values: [*][*:0]const u8 = undefined;
var count: c_int = 0;
const status = Status.fromInt(
nats_c.natsMsgHeader_Values(@ptrCast(self), key.ptr, @ptrCast(&values), &count),
);
// the user must use std.mem.spanTo on each item they want to read to get a
// slice, since we can't do that automatically without having to allocate.
return status.toError() orelse values[0..@intCast(count)];
}
// prefer using message.getHeaderIterator
pub fn getAllHeaderKeys(self: *Message) Error![][*:0]const u8 {
var keys: [*][*:0]const u8 = undefined;
var count: c_int = 0;
const status = Status.fromInt(nats_c.natsMsgHeader_Keys(@ptrCast(self), @ptrCast(&keys), &count));
// the user must use std.mem.spanTo on each item they want to read to get a
// slice, since we can't do that automatically without having to allocate.
// the returned slice
return status.toError() orelse keys[0..@intCast(count)];
}
pub const HeaderValueIterator = struct {
values: [][*:0]const u8,
index: usize = 0,
pub fn destroy(self: HeaderValueIterator) void {
std.heap.raw_c_allocator.free(self.values);
}
pub const deinit = HeaderValueIterator.destroy;
pub fn next(self: *HeaderValueIterator) ?[:0]const u8 {
if (self.index >= self.values.len) return null;
defer self.index += 1;
return std.mem.sliceTo(self.values[self.index], 0);
}
pub fn peek(self: *HeaderValueIterator) ?[:0]const u8 {
if (self.index >= self.values.len) return null;
return std.mem.sliceTo(self.values[self.index], 0);
}
};
pub const HeaderIterator = struct {
message: *Message,
keys: [][*:0]const u8,
index: usize = 0,
pub const ValueResolver = struct {
message: *Message,
key: [:0]const u8,
pub fn value(self: ValueResolver) Error![:0]const u8 {
// TODO: if we didn't care about the lifecycle of self.message, we
// could do catch unreachable here and make this error-free
return try self.message.getHeaderValue(self.key);
}
pub fn valueIterator(self: ValueResolver) Error!HeaderValueIterator {
return try self.message.getHeaderValueIterator(self.key);
}
};
pub fn destroy(self: *HeaderIterator) void {
std.heap.raw_c_allocator.free(self.keys);
}
pub const deinit = HeaderIterator.destroy;
pub fn next(self: *HeaderIterator) ?ValueResolver {
if (self.index >= self.keys.len) return null;
defer self.index += 1;
return .{
.message = self.message,
.key = std.mem.sliceTo(self.keys[self.index], 0),
};
}
pub fn peek(self: *HeaderIterator) ?ValueResolver {
if (self.index >= self.keys.len) return null;
return .{
.message = self.message,
.key = std.mem.sliceTo(self.keys[self.index], 0),
};
}
pub fn nextKey(self: *HeaderIterator) ?[:0]const u8 {
if (self.index >= self.keys.len) return null;
defer self.index += 1;
return std.mem.sliceTo(self.keys[self.index], 0);
}
pub fn peekKey(self: *HeaderIterator) ?[:0]const u8 {
if (self.index >= self.keys.len) return null;
return std.mem.sliceTo(self.keys[self.index], 0);
}
};
};
// TODO: not implementing jetstream API right now

View File

@ -28,19 +28,19 @@ const Status = err_.Status;
const thunk = @import("./thunk.zig");
pub const MessageCount = struct {
messages: c_int = 0,
bytes: c_int = 0,
};
pub const SubscriptionStats = struct {
pending: MessageCount = .{},
max_pending: MessageCount = .{},
delivered_messages: i64 = 0,
dropped_messages: i64 = 0,
};
pub const Subscription = opaque {
pub const MessageCount = struct {
messages: c_int = 0,
bytes: c_int = 0,
};
pub const SubscriptionStats = struct {
pending: MessageCount = .{},
max_pending: MessageCount = .{},
delivered_messages: i64 = 0,
dropped_messages: i64 = 0,
};
pub fn isValid(self: *Subscription) bool {
return nats_c.natsSubscription_IsValid(@ptrCast(self));
}
@ -84,7 +84,7 @@ pub const Subscription = opaque {
// invalid/closed subscriptions return null. should we convert that into an
// error? could return error.InvalidSubscription
const result = nats_c.natsSubscription_GetSubject(@ptrCast(self)) orelse return null;
return std.mem.spanTo(u8, result, 0);
return std.mem.sliceTo(result, 0);
}
pub fn setPendingLimits(self: *Subscription, limit: MessageCount) Error!void {

View File

@ -5,4 +5,5 @@ test {
_ = @import("./nats.zig");
_ = @import("./connection.zig");
_ = @import("./message.zig");
_ = @import("./subscription.zig");
}

View File

@ -5,56 +5,113 @@ const std = @import("std");
const nats = @import("nats");
test "message: create message" {
const subject = "hello";
const reply = "reply";
const data = "world";
const util = @import("./util.zig");
test "nats.Message" {
const message_subject: [:0]const u8 = "hello";
const message_reply: [:0]const u8 = "reply";
const message_data: [:0]const u8 = "world";
// have to initialize the library so the reference counter can correctly destroy
// objects, otherwise we segfault on trying to free the memory.
try nats.init(nats.default_spin_count);
defer nats.deinit();
const message = try nats.Message.create(subject, reply, data);
{
const message = try nats.Message.create(message_subject, null, message_data);
defer message.destroy();
}
{
const message = try nats.Message.create(message_subject, message_reply, null);
defer message.destroy();
}
{
const message = try nats.Message.create(message_subject, null, null);
defer message.destroy();
}
const message = try nats.Message.create(message_subject, message_reply, message_data);
defer message.destroy();
const message2 = try nats.Message.create(subject, null, data);
defer message2.destroy();
const subject = message.getSubject();
try std.testing.expectEqualStrings(message_subject, subject);
const message3 = try nats.Message.create(subject, data, null);
defer message3.destroy();
const reply = message.getReply() orelse return error.TestUnexpectedResult;
try std.testing.expectEqualStrings(message_reply, reply);
const message4 = try nats.Message.create(subject, null, null);
defer message4.destroy();
const data = message.getData() orelse return error.TestUnexpectedResult;
try std.testing.expectEqualStrings(message_data, data);
try std.testing.expectEqual(message_data.len, message.getDataLength());
const message_header: [:0]const u8 = "foo";
const message_hvalues: []const [:0]const u8 = &.{ "bar", "baz" };
try message.setHeaderValue(message_header, message_hvalues[0]);
try std.testing.expectEqualStrings(message_hvalues[0], try message.getHeaderValue(message_header));
try message.addHeaderValue(message_header, message_hvalues[1]);
try std.testing.expectEqualStrings(message_hvalues[0], try message.getHeaderValue(message_header));
{
var idx: usize = 0;
var val_iter = try message.getHeaderValueIterator(message_header);
defer val_iter.destroy();
while (val_iter.next()) |value| : (idx += 1) {
try std.testing.expect(idx < message_hvalues.len);
try std.testing.expectEqualStrings(message_hvalues[idx], value);
}
}
{
var header_iter = try message.getHeaderIterator();
defer header_iter.destroy();
while (header_iter.next()) |header| {
try std.testing.expectEqualStrings(message_header, header.key);
try std.testing.expectEqualStrings(message_hvalues[0], try header.value());
var idx: usize = 0;
var val_iter = try header.valueIterator();
defer val_iter.destroy();
while (val_iter.next()) |value| : (idx += 1) {
try std.testing.expect(idx < message_hvalues.len);
try std.testing.expectEqualStrings(message_hvalues[idx], value);
}
try std.testing.expect(val_iter.peek() == null);
}
try std.testing.expect(header_iter.peek() == null);
}
try message.deleteHeader(message_header);
_ = message.isNoResponders();
}
test "message: get subject" {
test "send nats.Message" {
var server = try util.TestServer.launch(.{});
defer server.stop();
try nats.init(nats.default_spin_count);
defer nats.deinit();
const subject = "hello";
const message = try nats.Message.create(subject, null, null);
const connection = try nats.Connection.connectTo(nats.default_server_url);
defer connection.destroy();
const message_subject: [:0]const u8 = "hello";
const message_reply: [:0]const u8 = "reply";
const message_data: [:0]const u8 = "world";
const message_header: [:0]const u8 = "foo";
const message_hvalues: []const [:0]const u8 = &.{ "bar", "baz" };
const message = try nats.Message.create(message_subject, message_reply, message_data);
defer message.destroy();
const received = message.getSubject();
try std.testing.expectEqualStrings(subject, received);
}
test "message: get reply" {
try nats.init(nats.default_spin_count);
defer nats.deinit();
const subject = "hello";
const reply = "reply";
const message = try nats.Message.create(subject, reply, null);
defer message.destroy();
const received = message.getReply() orelse return error.TestUnexpectedResult;
try std.testing.expectEqualStrings(reply, received);
const message2 = try nats.Message.create(subject, null, null);
defer message2.destroy();
const received2 = message2.getReply();
try std.testing.expect(received2 == null);
try message.setHeaderValue(message_header, message_hvalues[0]);
try message.addHeaderValue(message_header, message_hvalues[1]);
try connection.publishMessage(message);
}

180
tests/subscription.zig Normal file
View File

@ -0,0 +1,180 @@
// This file is licensed under the CC0 1.0 license.
// See: https://creativecommons.org/publicdomain/zero/1.0/legalcode
const std = @import("std");
const nats = @import("nats");
const util = @import("./util.zig");
test "nats.Subscription" {
var server = try util.TestServer.launch(.{});
defer server.stop();
try nats.init(nats.default_spin_count);
defer nats.deinit();
const connection = try nats.Connection.connectTo(nats.default_server_url);
defer connection.destroy();
const message_subject: [:0]const u8 = "hello";
const message_reply: [:0]const u8 = "reply";
const message_data: [:0]const u8 = "world";
const message = try nats.Message.create(message_subject, message_reply, message_data);
defer message.destroy();
{
const subscription = try connection.subscribeSync(message_subject);
defer subscription.destroy();
try subscription.autoUnsubscribe(1);
try subscription.setPendingLimits(.{ .messages = 10, .bytes = 1000 });
_ = try subscription.getPendingLimits();
_ = try subscription.getPending();
_ = try subscription.getMaxPending();
try subscription.clearMaxPending();
_ = try subscription.getDelivered();
_ = try subscription.getDropped();
_ = try subscription.getStats();
_ = try subscription.queuedMessageCount();
_ = subscription.getId();
const subj = subscription.getSubject() orelse return error.TestUnexpectedResult;
try std.testing.expectEqualStrings(message_subject, subj);
try connection.publishMessage(message);
const roundtrip = try subscription.nextMessage(1000);
try std.testing.expectEqualStrings(
message_data,
roundtrip.getData() orelse return error.TestUnexpectedResult,
);
try std.testing.expect(subscription.isValid() == false);
}
{
const subscription = try connection.queueSubscribeSync(message_subject, "queuegroup");
defer subscription.destroy();
try subscription.drain();
try subscription.waitForDrainCompletion(1000);
_ = subscription.drainCompletionStatus();
}
{
const subscription = try connection.subscribeSync(message_subject);
defer subscription.destroy();
try subscription.drain();
try subscription.waitForDrainCompletion(1000);
_ = subscription.drainCompletionStatus();
}
{
const subscription = try connection.subscribeSync(message_subject);
defer subscription.destroy();
try subscription.drainTimeout(1000);
try subscription.waitForDrainCompletion(1000);
}
}
fn onMessage(
userdata: *u32,
connection: *nats.Connection,
subscription: *nats.Subscription,
message: *nats.Message,
) void {
_ = subscription;
if (message.getReply()) |reply| {
connection.publish(reply, "greetings") catch @panic("OH NO");
} else @panic("HOW");
userdata.* += 1;
}
test "nats.Subscription (async)" {
var server = try util.TestServer.launch(.{});
defer server.stop();
try nats.init(nats.default_spin_count);
defer nats.deinit();
const connection = try nats.Connection.connectTo(nats.default_server_url);
defer connection.destroy();
const message_subject: [:0]const u8 = "hello";
const message_reply: [:0]const u8 = "reply";
const message_data: [:0]const u8 = "world";
const message = try nats.Message.create(message_subject, message_reply, message_data);
defer message.destroy();
{
var count: u32 = 0;
const subscription = try connection.subscribe(u32, message_subject, onMessage, &count);
defer subscription.destroy();
const response = try connection.requestMessage(message, 1000);
try std.testing.expectEqualStrings(
"greetings",
response.getData() orelse return error.TestUnexpectedResult,
);
}
{
var count: u32 = 0;
const subscription = try connection.subscribeTimeout(
u32,
message_subject,
1000,
onMessage,
&count,
);
defer subscription.destroy();
const response = try connection.requestMessage(message, 1000);
try std.testing.expectEqualStrings(
"greetings",
response.getData() orelse return error.TestUnexpectedResult,
);
}
{
var count: u32 = 0;
const subscription = try connection.queueSubscribe(
u32,
message_subject,
"queuegroup",
onMessage,
&count,
);
defer subscription.destroy();
const response = try connection.requestMessage(message, 1000);
try std.testing.expectEqualStrings(
"greetings",
response.getData() orelse return error.TestUnexpectedResult,
);
}
{
var count: u32 = 0;
const subscription = try connection.queueSubscribeTimeout(
u32,
message_subject,
"queuegroup",
1000,
onMessage,
&count,
);
defer subscription.destroy();
const response = try connection.requestMessage(message, 1000);
try std.testing.expectEqualStrings(
"greetings",
response.getData() orelse return error.TestUnexpectedResult,
);
}
}