factotum/nats.zig
1
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Compare commits

base: factotum:f742ee1a85efdccdf451e6b9b2ee7f0cc43e06c5
Branches Tags
factotum:master
factotum:zig-0.11.x
..
compare: factotum:b17a3fba6c2adaebda3f4e2862ef5f61be4740aa
Branches Tags
factotum:master
factotum:zig-0.11.x

No commits in common. "f742ee1a85efdccdf451e6b9b2ee7f0cc43e06c5" and "b17a3fba6c2adaebda3f4e2862ef5f61be4740aa" have entirely different histories.
f742ee1a85 ... b17a3fba6c

6 changed files with 54 additions and 70 deletions
Show Stats Expand all files Collapse all files

2
build.zig.zon
View File

@ -1,5 +1,5 @@
.{ .{
.name = "nats.zig", .name = "nats-client",
.version = "0.0.1", .version = "0.0.1",
.paths = .{ .paths = .{
"src", "src",

40
src/connection.zig
View File

@ -348,7 +348,7 @@ pub const Connection = opaque {
@ptrCast(self), @ptrCast(self),
subject.ptr, subject.ptr,
makeSubscriptionCallbackThunk(T, callback), makeSubscriptionCallbackThunk(T, callback),
@constCast(@ptrCast(userdata)), userdata,
)); ));
return status.toError() orelse sub; return status.toError() orelse sub;
} }
@ -369,7 +369,7 @@ pub const Connection = opaque {
subject.ptr, subject.ptr,
timeout, timeout,
makeSubscriptionCallbackThunk(T, callback), makeSubscriptionCallbackThunk(T, callback),
@constCast(@ptrCast(userdata)), userdata,
)); ));
return status.toError() orelse sub; return status.toError() orelse sub;
@ -403,7 +403,7 @@ pub const Connection = opaque {
subject.ptr, subject.ptr,
queue_group.ptr, queue_group.ptr,
makeSubscriptionCallbackThunk(T, callback), makeSubscriptionCallbackThunk(T, callback),
@constCast(@ptrCast(userdata)), userdata,
)); ));
return status.toError() orelse sub; return status.toError() orelse sub;
@ -427,7 +427,7 @@ pub const Connection = opaque {
queue_group.ptr, queue_group.ptr,
timeout, timeout,
makeSubscriptionCallbackThunk(T, callback), makeSubscriptionCallbackThunk(T, callback),
@constCast(@ptrCast(userdata)), userdata,
)); ));
return status.toError() orelse sub; return status.toError() orelse sub;
@ -504,7 +504,7 @@ pub const ConnectionOptions = opaque {
return Status.fromInt(nats_c.natsOptions_SetTokenHandler( return Status.fromInt(nats_c.natsOptions_SetTokenHandler(
@ptrCast(self), @ptrCast(self),
makeTokenCallbackThunk(T, callback), makeTokenCallbackThunk(T, callback),
@constCast(@ptrCast(userdata)), userdata,
)).raise(); )).raise();
} }
@ -644,7 +644,7 @@ pub const ConnectionOptions = opaque {
nats_c.natsOptions_SetCustomReconnectDelay( nats_c.natsOptions_SetCustomReconnectDelay(
@ptrCast(self), @ptrCast(self),
makeReconnectDelayCallbackThunk(T, callback), makeReconnectDelayCallbackThunk(T, callback),
@constCast(@ptrCast(userdata)), userdata,
), ),
).raise(); ).raise();
} }
@ -671,7 +671,7 @@ pub const ConnectionOptions = opaque {
nats_c.natsOptions_SetErrorHandler( nats_c.natsOptions_SetErrorHandler(
@ptrCast(self), @ptrCast(self),
makeErrorHandlerCallbackThunk(T, callback), makeErrorHandlerCallbackThunk(T, callback),
@constCast(@ptrCast(userdata)), userdata,
), ),
).raise(); ).raise();
} }
@ -685,7 +685,7 @@ pub const ConnectionOptions = opaque {
return Status.fromInt(nats_c.natsOptions_SetClosedCB( return Status.fromInt(nats_c.natsOptions_SetClosedCB(
@ptrCast(self), @ptrCast(self),
makeConnectionCallbackThunk(T, callback), makeConnectionCallbackThunk(T, callback),
@constCast(@ptrCast(userdata)), userdata,
)).raise(); )).raise();
} }
@ -698,7 +698,7 @@ pub const ConnectionOptions = opaque {
return Status.fromInt(nats_c.natsOptions_SetClosedCB( return Status.fromInt(nats_c.natsOptions_SetClosedCB(
@ptrCast(self), @ptrCast(self),
makeConnectionCallbackThunk(T, callback), makeConnectionCallbackThunk(T, callback),
@constCast(@ptrCast(userdata)), userdata,
)).raise(); )).raise();
} }
@ -711,7 +711,7 @@ pub const ConnectionOptions = opaque {
return Status.fromInt(nats_c.natsOptions_SetClosedCB( return Status.fromInt(nats_c.natsOptions_SetClosedCB(
@ptrCast(self), @ptrCast(self),
makeConnectionCallbackThunk(T, callback), makeConnectionCallbackThunk(T, callback),
@constCast(@ptrCast(userdata)), userdata,
)).raise(); )).raise();
} }
@ -724,7 +724,7 @@ pub const ConnectionOptions = opaque {
return Status.fromInt(nats_c.natsOptions_SetClosedCB( return Status.fromInt(nats_c.natsOptions_SetClosedCB(
@ptrCast(self), @ptrCast(self),
makeConnectionCallbackThunk(T, callback), makeConnectionCallbackThunk(T, callback),
@constCast(@ptrCast(userdata)), userdata,
)).raise(); )).raise();
} }
@ -737,7 +737,7 @@ pub const ConnectionOptions = opaque {
return Status.fromInt(nats_c.natsOptions_SetClosedCB( return Status.fromInt(nats_c.natsOptions_SetClosedCB(
@ptrCast(self), @ptrCast(self),
makeConnectionCallbackThunk(T, callback), makeConnectionCallbackThunk(T, callback),
@constCast(@ptrCast(userdata)), userdata,
)).raise(); )).raise();
} }
@ -753,7 +753,7 @@ pub const ConnectionOptions = opaque {
) Error!void { ) Error!void {
return Status.fromInt(nats_c.natsOptions_SetEventLoop( return Status.fromInt(nats_c.natsOptions_SetEventLoop(
@ptrCast(self), @ptrCast(self),
@constCast(@ptrCast(loop)), @ptrCast(loop),
makeAttachEventLoopCallbackThunk(T, L, attach_callback), makeAttachEventLoopCallbackThunk(T, L, attach_callback),
makeEventLoopAddRemoveCallbackThunk(T, read_callback), makeEventLoopAddRemoveCallbackThunk(T, read_callback),
makeEventLoopAddRemoveCallbackThunk(T, write_callback), makeEventLoopAddRemoveCallbackThunk(T, write_callback),
@ -822,7 +822,7 @@ pub const ConnectionOptions = opaque {
@ptrCast(self), @ptrCast(self),
retry, retry,
makeConnectionCallbackThunk(T, callback), makeConnectionCallbackThunk(T, callback),
@constCast(@ptrCast(userdata)), userdata,
)).raise(); )).raise();
} }
@ -838,17 +838,13 @@ pub const ConnectionOptions = opaque {
return Status.fromInt(nats_c.natsOptions_SetUserCredentialsCallbacks( return Status.fromInt(nats_c.natsOptions_SetUserCredentialsCallbacks(
@ptrCast(self), @ptrCast(self),
makeJwtHandlerCallbackThunk(T, jwt_callback), makeJwtHandlerCallbackThunk(T, jwt_callback),
@constCast(@ptrCast(jwt_userdata)), jwt_userdata,
makeSignatureHandlerCallbackThunk(U, sig_callback), makeSignatureHandlerCallbackThunk(U, sig_callback),
@constCast(@ptrCast(sig_userdata)), sig_userdata,
)).raise(); )).raise();
} }
pub fn setUserCredentialsFromFiles( pub fn setUserCredentialsFromFiles(self: *ConnectionOptions, user_or_chained_file: [:0]const u8, seed_file: [:0]const u8) Error!void {
self: *ConnectionOptions,
user_or_chained_file: [:0]const u8,
seed_file: [:0]const u8,
) Error!void {
return Status.fromInt( return Status.fromInt(
nats_c.natsOptions_SetUserCredentialsFromFiles( nats_c.natsOptions_SetUserCredentialsFromFiles(
@ptrCast(self), @ptrCast(self),
@ -878,7 +874,7 @@ pub const ConnectionOptions = opaque {
@ptrCast(self), @ptrCast(self),
pub_key.ptr, pub_key.ptr,
makeSignatureHandlerCallbackThunk(T, sig_callback), makeSignatureHandlerCallbackThunk(T, sig_callback),
@constCast(@ptrCast(sig_userdata)), sig_userdata,
)).raise(); )).raise();
} }

6
src/subscription.zig
View File

@ -174,12 +174,12 @@ pub const Subscription = opaque {
self: *Subscription, self: *Subscription,
comptime T: type, comptime T: type,
comptime callback: *const thunk.SimpleCallbackThunkSignature(T), comptime callback: *const thunk.SimpleCallbackThunkSignature(T),
userdata: T, userdata: *T,
) Error!void { ) Error!void {
return Status.fromInt(nats_c.natsSubscription_SetOnCompleteCB( return Status.fromInt(nats_c.natsSubscription_SetOnCompleteCB(
@ptrCast(self), @ptrCast(self),
thunk.makeSimpleCallbackThunk(T, callback), thunk.makeSimpleCallbackThunk(callback),
@constCast(@ptrCast(userdata)), userdata,
)).raise(); )).raise();
} }
}; };

6
src/thunk.zig
View File

@ -23,7 +23,7 @@ pub fn checkUserDataType(comptime T: type) void {
.Optional => |info| switch (@typeInfo(info.child)) { .Optional => |info| switch (@typeInfo(info.child)) {
.Optional => @compileError( .Optional => @compileError(
"nats callbacks can only accept an (optional) single, many," ++ "nats callbacks can only accept an (optional) single, many," ++
" or c pointer as userdata. \"" ++ " or c pointer as userdata, not slices. \"" ++
@typeName(T) ++ "\" has more than one optional specifier.", @typeName(T) ++ "\" has more than one optional specifier.",
), ),
else => checkUserDataType(info.child), else => checkUserDataType(info.child),
@ -38,7 +38,7 @@ pub fn checkUserDataType(comptime T: type) void {
}, },
else => @compileError( else => @compileError(
"nats callbacks can only accept an (optional) single, many," ++ "nats callbacks can only accept an (optional) single, many," ++
" or c pointer as userdata. \"" ++ " or c pointer as userdata, not slices. \"" ++
@typeName(T) ++ "\" is not a pointer type.", @typeName(T) ++ "\" is not a pointer type.",
), ),
} }
@ -57,7 +57,7 @@ pub fn makeSimpleCallbackThunk(
comptime checkUserDataType(T); comptime checkUserDataType(T);
return struct { return struct {
fn thunk(userdata: ?*anyopaque) callconv(.C) void { fn thunk(userdata: ?*anyopaque) callconv(.C) void {
const data: T = if (userdata) |u| @alignCast(@ptrCast(u)) else unreachable; const data: *T = if (userdata) |u| @alignCast(@ptrCast(u)) else unreachable;
callback(data); callback(data);
} }
}.thunk; }.thunk;

28
tests/connection.zig
View File

@ -99,7 +99,7 @@ test "nats.Connection" {
connection.drainTimeout(1000) catch {}; connection.drainTimeout(1000) catch {};
} }
fn reconnectDelayHandler(userdata: *const u32, connection: *nats.Connection, attempts: c_int) i64 { fn reconnectDelayHandler(userdata: *u32, connection: *nats.Connection, attempts: c_int) i64 {
_ = userdata; _ = userdata;
_ = connection; _ = connection;
_ = attempts; _ = attempts;
@ -108,7 +108,7 @@ fn reconnectDelayHandler(userdata: *const u32, connection: *nats.Connection, att
} }
fn errorHandler( fn errorHandler(
userdata: *const u32, userdata: *u32,
connection: *nats.Connection, connection: *nats.Connection,
subscription: *nats.Subscription, subscription: *nats.Subscription,
status: nats.Status, status: nats.Status,
@ -119,18 +119,18 @@ fn errorHandler(
_ = status; _ = status;
} }
fn connectionHandler(userdata: *const u32, connection: *nats.Connection) void { fn connectionHandler(userdata: *u32, connection: *nats.Connection) void {
_ = userdata; _ = userdata;
_ = connection; _ = connection;
} }
fn jwtHandler(userdata: *const u32) nats.JwtResponseOrError { fn jwtHandler(userdata: *u32) nats.JwtResponseOrError {
_ = userdata; _ = userdata;
// return .{ .jwt = std.heap.raw_c_allocator.dupeZ(u8, "abcdef") catch @panic("no!") }; // return .{ .jwt = std.heap.raw_c_allocator.dupeZ(u8, "abcdef") catch @panic("no!") };
return .{ .error_message = std.heap.raw_c_allocator.dupeZ(u8, "dang") catch @panic("no!") }; return .{ .error_message = std.heap.raw_c_allocator.dupeZ(u8, "dang") catch @panic("no!") };
} }
fn signatureHandler(userdata: *const u32, nonce: [:0]const u8) nats.SignatureResponseOrError { fn signatureHandler(userdata: *u32, nonce: [:0]const u8) nats.SignatureResponseOrError {
_ = userdata; _ = userdata;
_ = nonce; _ = nonce;
// return .{ .signature = std.heap.raw_c_allocator.dupe(u8, "01230123") catch @panic("no!") }; // return .{ .signature = std.heap.raw_c_allocator.dupe(u8, "01230123") catch @panic("no!") };
@ -144,7 +144,7 @@ test "nats.ConnectionOptions" {
const options = try nats.ConnectionOptions.create(); const options = try nats.ConnectionOptions.create();
defer options.destroy(); defer options.destroy();
const userdata: u32 = 0; var userdata: u32 = 0;
try options.setUrl(nats.default_server_url); try options.setUrl(nats.default_server_url);
const servers = [_][*:0]const u8{ "nats://127.0.0.1:4442", "nats://127.0.0.1:4443" }; const servers = [_][*:0]const u8{ "nats://127.0.0.1:4442", "nats://127.0.0.1:4443" };
@ -164,14 +164,14 @@ test "nats.ConnectionOptions" {
try options.setMaxReconnect(10); try options.setMaxReconnect(10);
try options.setReconnectWait(500); try options.setReconnectWait(500);
try options.setReconnectJitter(100, 200); try options.setReconnectJitter(100, 200);
try options.setCustomReconnectDelay(*const u32, reconnectDelayHandler, &userdata); try options.setCustomReconnectDelay(*u32, reconnectDelayHandler, &userdata);
try options.setReconnectBufSize(1024); try options.setReconnectBufSize(1024);
try options.setMaxPendingMessages(50); try options.setMaxPendingMessages(50);
try options.setErrorHandler(*const u32, errorHandler, &userdata); try options.setErrorHandler(*u32, errorHandler, &userdata);
try options.setClosedCallback(*const u32, connectionHandler, &userdata); try options.setClosedCallback(*u32, connectionHandler, &userdata);
try options.setDisconnectedCallback(*const u32, connectionHandler, &userdata); try options.setDisconnectedCallback(*u32, connectionHandler, &userdata);
try options.setDiscoveredServersCallback(*const u32, connectionHandler, &userdata); try options.setDiscoveredServersCallback(*u32, connectionHandler, &userdata);
try options.setLameDuckModeCallback(*const u32, connectionHandler, &userdata); try options.setLameDuckModeCallback(*u32, connectionHandler, &userdata);
try options.ignoreDiscoveredServers(true); try options.ignoreDiscoveredServers(true);
try options.useGlobalMessageDelivery(false); try options.useGlobalMessageDelivery(false);
try options.ipResolutionOrder(.ipv4_first); try options.ipResolutionOrder(.ipv4_first);
@ -179,8 +179,8 @@ test "nats.ConnectionOptions" {
try options.useOldRequestStyle(false); try options.useOldRequestStyle(false);
try options.setFailRequestsOnDisconnect(true); try options.setFailRequestsOnDisconnect(true);
try options.setNoEcho(true); try options.setNoEcho(true);
try options.setRetryOnFailedConnect(*const u32, connectionHandler, true, &userdata); try options.setRetryOnFailedConnect(*u32, connectionHandler, true, &userdata);
try options.setUserCredentialsCallbacks(*const u32, *const u32, jwtHandler, signatureHandler, &userdata, &userdata); try options.setUserCredentialsCallbacks(*u32, *u32, jwtHandler, signatureHandler, &userdata, &userdata);
try options.setWriteDeadline(5); try options.setWriteDeadline(5);
try options.disableNoResponders(true); try options.disableNoResponders(true);
try options.setCustomInboxPrefix("_FOOBOX"); try options.setCustomInboxPrefix("_FOOBOX");

30
tests/subscription.zig
View File

@ -80,21 +80,18 @@ test "nats.Subscription" {
} }
fn onMessage( fn onMessage(
userdata: *const u32, userdata: *u32,
connection: *nats.Connection, connection: *nats.Connection,
subscription: *nats.Subscription, subscription: *nats.Subscription,
message: *nats.Message, message: *nats.Message,
) void { ) void {
_ = subscription; _ = subscription;
_ = userdata;
if (message.getReply()) |reply| { if (message.getReply()) |reply| {
connection.publish(reply, "greetings") catch @panic("OH NO"); connection.publish(reply, "greetings") catch @panic("OH NO");
} else @panic("HOW"); } else @panic("HOW");
}
fn onClose(userdata: *[]const u8) void { userdata.* += 1;
userdata.* = "closed";
} }
test "nats.Subscription (async)" { test "nats.Subscription (async)" {
@ -115,30 +112,21 @@ test "nats.Subscription (async)" {
defer message.destroy(); defer message.destroy();
{ {
var closed: []const u8 = "test"; var count: u32 = 0;
{ const subscription = try connection.subscribe(*u32, message_subject, onMessage, &count);
const count: u32 = 0;
const subscription = try connection.subscribe(*const u32, message_subject, onMessage, &count);
defer subscription.destroy(); defer subscription.destroy();
try subscription.setCompletionCallback(*[]const u8, onClose, &closed);
const response = try connection.requestMessage(message, 1000); const response = try connection.requestMessage(message, 1000);
try std.testing.expectEqualStrings( try std.testing.expectEqualStrings(
"greetings", "greetings",
response.getData() orelse return error.TestUnexpectedResult, response.getData() orelse return error.TestUnexpectedResult,
); );
} }
// we have to sleep to allow the close callback to run. I am worried this may
// still end up being flaky, however.
nats.sleep(1);
try std.testing.expectEqualStrings("closed", closed);
}
{ {
const count: u32 = 0; var count: u32 = 0;
const subscription = try connection.subscribeTimeout( const subscription = try connection.subscribeTimeout(
*const u32, *u32,
message_subject, message_subject,
1000, 1000,
onMessage, onMessage,
@ -154,9 +142,9 @@ test "nats.Subscription (async)" {
} }
{ {
const count: u32 = 0; var count: u32 = 0;
const subscription = try connection.queueSubscribe( const subscription = try connection.queueSubscribe(
*const u32, *u32,
message_subject, message_subject,
"queuegroup", "queuegroup",
onMessage, onMessage,
@ -174,7 +162,7 @@ test "nats.Subscription (async)" {
{ {
var count: u32 = 0; var count: u32 = 0;
const subscription = try connection.queueSubscribeTimeout( const subscription = try connection.queueSubscribeTimeout(
*const u32, *u32,
message_subject, message_subject,
"queuegroup", "queuegroup",
1000, 1000,