yuzucchii/jetzig
1
0
Fork 0
mirror of https://github.com/jetzig-framework/jetzig.git synced 2025-05-14 22:16:08 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #56 from jetzig-framework/background-jobs

Browse Source 
Background jobs
This commit is contained in:
bobf 2024-04-15 21:06:48 +01:00 committed by GitHub
commit ba0e488894
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
27 changed files with 838 additions and 115 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

22
build.zig
View File

@ -1,12 +1,15 @@
const std = @import("std");
pub const GenerateRoutes = @import("src/GenerateRoutes.zig");
pub const Routes = @import("src/Routes.zig");
pub const GenerateMimeTypes = @import("src/GenerateMimeTypes.zig");
pub const TemplateFn = @import("src/jetzig.zig").TemplateFn;
pub const StaticRequest = @import("src/jetzig.zig").StaticRequest;
pub const http = @import("src/jetzig/http.zig");
pub const data = @import("src/jetzig/data.zig");
pub const views = @import("src/jetzig/views.zig");
pub const Route = views.Route;
pub const Job = @import("src/jetzig.zig").Job;
const zmpl_build = @import("zmpl");
pub fn build(b: *std.Build) !void {
@ -72,6 +75,8 @@ pub fn build(b: *std.Build) !void {
const zmpl_module = zmpl_dep.module("zmpl");
const jetkv_dep = b.dependency("jetkv", .{ .target = target, .optimize = optimize });
// This is the way to make it look nice in the zig build script
// If we would do it the other way around, we would have to do
// b.dependency("jetzig",.{}).builder.dependency("zmpl",.{}).module("zmpl");
@ -83,6 +88,7 @@ pub fn build(b: *std.Build) !void {
jetzig_module.addImport("zmpl", zmpl_module);
jetzig_module.addImport("args", zig_args_dep.module("args"));
jetzig_module.addImport("zmd", zmd_dep.module("zmd"));
jetzig_module.addImport("jetkv", jetkv_dep.module("jetkv"));
const main_tests = b.addTest(.{
.root_source_file = .{ .path = "src/tests.zig" },
@ -100,6 +106,7 @@ pub fn build(b: *std.Build) !void {
docs_step.dependOn(&docs_install.step);
main_tests.root_module.addImport("zmpl", zmpl_dep.module("zmpl"));
main_tests.root_module.addImport("jetkv", jetkv_dep.module("jetkv"));
const run_main_tests = b.addRunArtifact(main_tests);
const test_step = b.step("test", "Run library tests");
@ -132,7 +139,18 @@ pub fn jetzigInit(b: *std.Build, exe: *std.Build.Step.Compile, options: JetzigIn
}
}
var generate_routes = try GenerateRoutes.init(b.allocator, "src/app/views");
const root_path = b.build_root.path orelse try std.fs.cwd().realpathAlloc(b.allocator, ".");
const templates_path = try std.fs.path.join(
b.allocator,
&[_][]const u8{ root_path, "src", "app", "views" },
);
const jobs_path = try std.fs.path.join(
b.allocator,
&[_][]const u8{ root_path, "src", "app", "jobs" },
);
var generate_routes = try Routes.init(b.allocator, root_path, templates_path, jobs_path);
try generate_routes.generateRoutes();
const write_files = b.addWriteFiles();
const routes_file = write_files.add("routes.zig", generate_routes.buffer.items);

4
build.zig.zon
View File

@ -14,6 +14,10 @@
.url = "https://github.com/MasterQ32/zig-args/archive/01d72b9a0128c474aeeb9019edd48605fa6d95f7.tar.gz",
.hash = "12208a1de366740d11de525db7289345949f5fd46527db3f89eecc7bb49b012c0732",
},
.jetkv = .{
.url = "https://github.com/jetzig-framework/jetkv/archive/a6fcc2df220c1a40094e167eeb567bb5888404e9.tar.gz",
.hash = "12207bd2d7465b33e745a5b0567172377f94a221d1fc9aab238bb1b372c64f4ec1a0",
},
},
.paths = .{

43
cli/commands/generate.zig
View File

@ -4,27 +4,21 @@ const view = @import("generate/view.zig");
const partial = @import("generate/partial.zig");
const layout = @import("generate/layout.zig");
const middleware = @import("generate/middleware.zig");
const job = @import("generate/job.zig");
const secret = @import("generate/secret.zig");
const util = @import("../util.zig");
/// Command line options for the `generate` command.
pub const Options = struct {
pub const meta = .{
.usage_summary = "[view|partial|layout|middleware|secret] [options]",
.usage_summary = "[view|partial|layout|middleware|job|secret] [options]",
.full_text =
\\Generates scaffolding for views, middleware, and other objects in future.
\\Generate scaffolding for views, middleware, and other objects.
\\
\\When generating a view, by default all actions will be included.
\\Optionally pass one or more of the following arguments to specify desired actions:
\\Pass `--help` to any generator for more information, e.g.:
\\
\\ index, get, post, patch, put, delete
\\ jetzig generate view --help
\\
\\Each view action can be qualified with a `:static` option to mark the view content
\\as statically generated at build time.
\\
\\e.g. generate a view named `iguanas` with a static `index` action:
\\
\\ jetzig generate view iguanas index:static get post delete
,
};
};
@ -41,11 +35,8 @@ pub fn run(
defer cwd.close();
_ = options;
if (other_options.help) {
try args.printHelp(Options, "jetzig generate", writer);
return;
}
var generate_type: ?enum { view, partial, layout, middleware, secret } = null;
var generate_type: ?enum { view, partial, layout, middleware, job, secret } = null;
var sub_args = std.ArrayList([]const u8).init(allocator);
defer sub_args.deinit();
@ -56,6 +47,8 @@ pub fn run(
generate_type = .partial;
} else if (generate_type == null and std.mem.eql(u8, arg, "layout")) {
generate_type = .layout;
} else if (generate_type == null and std.mem.eql(u8, arg, "job")) {
generate_type = .job;
} else if (generate_type == null and std.mem.eql(u8, arg, "middleware")) {
generate_type = .middleware;
} else if (generate_type == null and std.mem.eql(u8, arg, "secret")) {
@ -68,16 +61,22 @@ pub fn run(
}
}
if (other_options.help and generate_type == null) {
try args.printHelp(Options, "jetzig generate", writer);
return;
}
if (generate_type) |capture| {
return switch (capture) {
.view => view.run(allocator, cwd, sub_args.items),
.partial => partial.run(allocator, cwd, sub_args.items),
.layout => layout.run(allocator, cwd, sub_args.items),
.middleware => middleware.run(allocator, cwd, sub_args.items),
.secret => secret.run(allocator, cwd, sub_args.items),
.view => view.run(allocator, cwd, sub_args.items, other_options.help),
.partial => partial.run(allocator, cwd, sub_args.items, other_options.help),
.layout => layout.run(allocator, cwd, sub_args.items, other_options.help),
.job => job.run(allocator, cwd, sub_args.items, other_options.help),
.middleware => middleware.run(allocator, cwd, sub_args.items, other_options.help),
.secret => secret.run(allocator, cwd, sub_args.items, other_options.help),
};
} else {
std.debug.print("Missing sub-command. Expected: [view|partial|layout|middleware]\n", .{});
std.debug.print("Missing sub-command. Expected: [view|partial|layout|job|middleware|secret]\n", .{});
return error.JetzigCommandError;
}
}

58
cli/commands/generate/job.zig Normal file
View File

@ -0,0 +1,58 @@
const std = @import("std");
/// Run the job generator. Create a job in `src/app/jobs/`
pub fn run(allocator: std.mem.Allocator, cwd: std.fs.Dir, args: [][]const u8, help: bool) !void {
if (help or args.len != 1) {
std.debug.print(
\\Generate a new Job. Jobs can be scheduled to run in the background.
\\Use a Job when you need to return a request immediately and perform
\\another task asynchronously.
\\
\\Example:
\\
\\ jetzig generate job iguana
\\
, .{});
if (help) return;
return error.JetzigCommandError;
}
const dir_path = try std.fs.path.join(allocator, &[_][]const u8{ "src", "app", "jobs" });
defer allocator.free(dir_path);
var dir = try cwd.makeOpenPath(dir_path, .{});
defer dir.close();
const filename = try std.mem.concat(allocator, u8, &[_][]const u8{ args[0], ".zig" });
defer allocator.free(filename);
const file = dir.createFile(filename, .{ .exclusive = true }) catch |err| {
switch (err) {
error.PathAlreadyExists => {
std.debug.print("Partial already exists: {s}\n", .{filename});
return error.JetzigCommandError;
},
else => return err,
}
};
try file.writeAll(
\\const std = @import("std");
\\const jetzig = @import("jetzig");
\\
\\/// The `run` function for all jobs receives an arena allocator, a logger, and the params
\\/// passed to the job when it was created.
\\pub fn run(allocator: std.mem.Allocator, params: *jetzig.data.Value, logger: jetzig.Logger) !void {
\\ // Job execution code
\\}
\\
);
file.close();
const realpath = try dir.realpathAlloc(allocator, filename);
defer allocator.free(realpath);
std.debug.print("Generated job: {s}\n", .{realpath});
}

11
cli/commands/generate/layout.zig
View File

@ -1,16 +1,21 @@
const std = @import("std");
/// Run the layout generator. Create a layout template in `src/app/views/layouts`
pub fn run(allocator: std.mem.Allocator, cwd: std.fs.Dir, args: [][]const u8) !void {
if (args.len != 1) {
pub fn run(allocator: std.mem.Allocator, cwd: std.fs.Dir, args: [][]const u8, help: bool) !void {
if (help or args.len != 1) {
std.debug.print(
\\Expected a layout name.
\\Generate a layout. Layouts encapsulate common boilerplate mark-up.
\\
\\Specify a layout name to create a new Zmpl template in src/app/views/layouts/
\\
\\Example:
\\
\\ jetzig generate layout standard
\\
, .{});
if (help) return;
return error.JetzigCommandError;
}

9
cli/commands/generate/middleware.zig
View File

@ -2,16 +2,19 @@ const std = @import("std");
const util = @import("../../util.zig");
/// Run the middleware generator. Create a middleware file in `src/app/middleware/`
pub fn run(allocator: std.mem.Allocator, cwd: std.fs.Dir, args: [][]const u8) !void {
if (args.len != 1 or !util.isCamelCase(args[0])) {
pub fn run(allocator: std.mem.Allocator, cwd: std.fs.Dir, args: [][]const u8, help: bool) !void {
if (help or args.len != 1 or !util.isCamelCase(args[0])) {
std.debug.print(
\\Expected a middleware name in CamelCase.
\\Generate a middleware module. Module name must be in CamelCase.
\\
\\Example:
\\
\\ jetzig generate middleware IguanaBrain
\\
, .{});
if (help) return;
return error.JetzigCommandError;
}

9
cli/commands/generate/partial.zig
View File

@ -1,16 +1,19 @@
const std = @import("std");
/// Run the partial generator. Create a partial template in `src/app/views/`
pub fn run(allocator: std.mem.Allocator, cwd: std.fs.Dir, args: [][]const u8) !void {
if (args.len != 2) {
pub fn run(allocator: std.mem.Allocator, cwd: std.fs.Dir, args: [][]const u8, help: bool) !void {
if (help or args.len != 2) {
std.debug.print(
\\Expected a view name and a name for a partial.
\\Generate a partial template. Expects a view name followed by a partial name.
\\
\\Example:
\\
\\ jetzig generate partial iguanas ziglet
\\
, .{});
if (help) return;
return error.JetzigCommandError;
}

10
cli/commands/generate/secret.zig
View File

@ -1,7 +1,15 @@
const std = @import("std");
/// Generate a secure random secret and output to stdout.
pub fn run(allocator: std.mem.Allocator, cwd: std.fs.Dir, args: [][]const u8) !void {
pub fn run(allocator: std.mem.Allocator, cwd: std.fs.Dir, args: [][]const u8, help: bool) !void {
if (help) {
std.debug.print(
\\Generate a secure random secret suitable for use as the `JETZIG_SECRET` environment variable.
\\
, .{});
return;
}
_ = allocator;
_ = args;
_ = cwd;

18
cli/commands/generate/view.zig
View File

@ -2,17 +2,27 @@ const std = @import("std");
const util = @import("../../util.zig");
/// Run the view generator. Create a view in `src/app/views/`
pub fn run(allocator: std.mem.Allocator, cwd: std.fs.Dir, args: [][]const u8) !void {
if (args.len == 0) {
std.debug.print(".\n", .{});
pub fn run(allocator: std.mem.Allocator, cwd: std.fs.Dir, args: [][]const u8, help: bool) !void {
if (help or args.len == 0) {
std.debug.print(
\\Expected view name followed by optional actions.
\\Generate a view. Pass optional action names from:
\\ index, get, post, put, patch, delete
\\
\\Optionally suffix actions with `:static` to use static routing.
\\Static requests are rendered at build time only. Use static routes
\\when rendering takes a long time and content does not change between
\\deployments.
\\
\\Omit action names to generate a view with all actions defined.
\\
\\Example:
\\
\\ jetzig generate view iguanas index:static get post delete
\\
, .{});
if (help) return;
return error.JetzigCommandError;
}

9
demo/src/app/jobs/example.zig Normal file
View File

@ -0,0 +1,9 @@
const std = @import("std");
const jetzig = @import("jetzig");
/// The `run` function for all jobs receives an arena allocator, a logger, and the params
/// passed to the job when it was created.
pub fn run(allocator: std.mem.Allocator, params: *jetzig.data.Value, logger: jetzig.Logger) !void {
_ = allocator;
try logger.INFO("Job received params: {s}", .{try params.toJson()});
}

8
demo/src/app/jobs/iguana.zig Normal file
View File

@ -0,0 +1,8 @@
const std = @import("std");
const jetzig = @import("jetzig");
/// The `run` function for all jobs receives an arena allocator, a logger, and the params
/// passed to the job when it was created.
pub fn run(allocator: std.mem.Allocator, params: *jetzig.data.Value, logger: jetzig.Logger) !void {
// Job execution code
}

21
demo/src/app/views/background_jobs.zig Normal file
View File

@ -0,0 +1,21 @@
const std = @import("std");
const jetzig = @import("jetzig");
/// This example demonstrates usage of Jetzig's background jobs.
pub fn index(request: *jetzig.Request, data: *jetzig.Data) !jetzig.View {
// Create a new job using `src/app/jobs/example_job.zig`.
var job = try request.job("example");
// Add a param `foo` to the job.
try job.put("foo", data.string("bar"));
try job.put("id", data.integer(std.crypto.random.int(u32)));
// Schedule the job for background processing. The job is added to the queue. When the job is
// processed a new instance of `example_job` is created and its `run` function is invoked.
// All params are added above are available to the job by calling `job.params()` inside the
// `run` function.
try job.schedule();
return request.render(.ok);
}

35
demo/src/app/views/kvstore.zig Normal file
View File

@ -0,0 +1,35 @@
const jetzig = @import("jetzig");
/// This example demonstrates usage of Jetzig's KV store.
pub fn index(request: *jetzig.Request, data: *jetzig.Data) !jetzig.View {
var root = try data.object();
// Fetch a string from the KV store. If it exists, store it in the root data object,
// otherwise store a string value to be picked up by the next request.
if (request.kvGet(.string, "example-key")) |capture| {
try root.put("stored_string", data.string(capture));
} else {
try request.kvPut(.string, "example-key", "example-value");
}
// Pop an item from the array and store it in the root data object. This will empty the
// array after multiple requests.
if (request.kvPop("example-array")) |string| try root.put("popped", data.string(string));
// Fetch an array from the KV store. If it exists, store its values in the root data object,
// otherwise store a new array to be picked up by the next request.
if (request.kvGet(.array, "example-array")) |kv_array| {
var array = try data.array();
for (kv_array.items()) |item| try array.append(data.string(item));
try root.put("stored_array", array);
} else {
// Create a KV Array and store it in the key value store.
var kv_array = request.kvArray();
try kv_array.append("hello");
try kv_array.append("goodbye");
try kv_array.append("hello again");
try request.kvPut(.array, "example-array", kv_array);
}
return request.render(.ok);
}

8
demo/src/main.zig
View File

@ -32,6 +32,14 @@ pub const jetzig_options = struct {
// HTTP buffer. Must be large enough to store all headers. This should typically not be modified.
// pub const http_buffer_size: usize = std.math.pow(usize, 2, 16);
// The number of worker threads to spawn on startup for processing Jobs (NOT the number of
// HTTP server worker threads).
pub const job_worker_threads: usize = 4;
// Duration before looking for more Jobs when the queue is found to be empty, in
// milliseconds.
// pub const job_worker_sleep_interval_ms: usize = 10;
// Set custom fragments for rendering markdown templates. Any values will fall back to
// defaults provided by Zmd (https://github.com/bobf/zmd/blob/main/src/zmd/html.zig).
pub const markdown_fragments = struct {

250
src/GenerateRoutes.zig → src/Routes.zig
View File

@ -3,18 +3,21 @@ const jetzig = @import("jetzig.zig");
ast: std.zig.Ast = undefined,
allocator: std.mem.Allocator,
root_path: []const u8,
views_path: []const u8,
jobs_path: []const u8,
buffer: std.ArrayList(u8),
dynamic_routes: std.ArrayList(Function),
static_routes: std.ArrayList(Function),
data: *jetzig.data.Data,
const Self = @This();
const Routes = @This();
const Function = struct {
name: []const u8,
view_name: []const u8,
args: []Arg,
routes: *const Routes,
path: []const u8,
source: []const u8,
params: std.ArrayList([]const u8),
@ -23,31 +26,36 @@ const Function = struct {
/// The full name of a route. This **must** match the naming convention used by static route
/// compilation.
/// path: `src/app/views/iguanas.zig`, action: `index` => `iguanas_index`
pub fn fullName(self: @This(), allocator: std.mem.Allocator) ![]const u8 {
const relative_path = try std.fs.path.relative(allocator, "src/app/views/", self.path);
defer allocator.free(relative_path);
pub fn fullName(self: Function) ![]const u8 {
const relative_path = try self.routes.relativePathFrom(.views, self.path, .posix);
defer self.routes.allocator.free(relative_path);
const path = relative_path[0 .. relative_path.len - std.fs.path.extension(relative_path).len];
std.mem.replaceScalar(u8, path, '\\', '/');
std.mem.replaceScalar(u8, path, '/', '_');
return std.mem.concat(allocator, u8, &[_][]const u8{ path, "_", self.name });
return std.mem.concat(self.routes.allocator, u8, &[_][]const u8{ path, "_", self.name });
}
pub fn viewName(self: Function) ![]const u8 {
const relative_path = try self.routes.relativePathFrom(.views, self.path, .posix);
defer self.routes.allocator.free(relative_path);
return try self.routes.allocator.dupe(u8, chompExtension(relative_path));
}
/// The path used to match the route. Resource ID and extension is not included here and is
/// appended as needed during matching logic at run time.
pub fn uriPath(self: @This(), allocator: std.mem.Allocator) ![]const u8 {
const relative_path = try std.fs.path.relative(allocator, "src/app/views/", self.path);
defer allocator.free(relative_path);
pub fn uriPath(self: Function) ![]const u8 {
const relative_path = try self.routes.relativePathFrom(.views, self.path, .posix);
defer self.routes.allocator.free(relative_path);
const path = relative_path[0 .. relative_path.len - std.fs.path.extension(relative_path).len];
std.mem.replaceScalar(u8, path, '\\', '/');
if (std.mem.eql(u8, path, "root")) return try allocator.dupe(u8, "/");
if (std.mem.eql(u8, path, "root")) return try self.routes.allocator.dupe(u8, "/");
return try std.mem.concat(allocator, u8, &[_][]const u8{ "/", path });
return try std.mem.concat(self.routes.allocator, u8, &[_][]const u8{ "/", path });
}
pub fn lessThanFn(context: void, lhs: @This(), rhs: @This()) bool {
pub fn lessThanFn(context: void, lhs: Function, rhs: Function) bool {
_ = context;
return std.mem.order(u8, lhs.name, rhs.name).compare(std.math.CompareOperator.lt);
}
@ -58,7 +66,7 @@ const Arg = struct {
name: []const u8,
type_name: []const u8,
pub fn typeBasename(self: @This()) ![]const u8 {
pub fn typeBasename(self: Arg) ![]const u8 {
if (std.mem.indexOfScalar(u8, self.type_name, '.')) |_| {
var it = std.mem.splitBackwardsScalar(u8, self.type_name, '.');
while (it.next()) |capture| {
@ -76,13 +84,20 @@ const Arg = struct {
}
};
pub fn init(allocator: std.mem.Allocator, views_path: []const u8) !Self {
pub fn init(
allocator: std.mem.Allocator,
root_path: []const u8,
views_path: []const u8,
jobs_path: []const u8,
) !Routes {
const data = try allocator.create(jetzig.data.Data);
data.* = jetzig.data.Data.init(allocator);
return .{
.allocator = allocator,
.root_path = root_path,
.views_path = views_path,
.jobs_path = jobs_path,
.buffer = std.ArrayList(u8).init(allocator),
.static_routes = std.ArrayList(Function).init(allocator),
.dynamic_routes = std.ArrayList(Function).init(allocator),
@ -90,7 +105,7 @@ pub fn init(allocator: std.mem.Allocator, views_path: []const u8) !Self {
};
}
pub fn deinit(self: *Self) void {
pub fn deinit(self: *Routes) void {
self.ast.deinit(self.allocator);
self.buffer.deinit();
self.static_routes.deinit();
@ -98,13 +113,76 @@ pub fn deinit(self: *Self) void {
}
/// Generates the complete route set for the application
pub fn generateRoutes(self: *Self) !void {
pub fn generateRoutes(self: *Routes) !void {
const writer = self.buffer.writer();
var views_dir = try std.fs.cwd().openDir(self.views_path, .{ .iterate = true });
defer views_dir.close();
try writer.writeAll(
\\const jetzig = @import("jetzig");
\\
\\pub const routes = [_]jetzig.Route{
\\
);
var walker = try views_dir.walk(self.allocator);
try self.writeRoutes(writer);
try writer.writeAll(
\\};
\\
);
try writer.writeAll(
\\
\\pub const jobs = [_]jetzig.JobDefinition{
\\
);
try self.writeJobs(writer);
try writer.writeAll(
\\};
\\
);
// std.debug.print("routes.zig\n{s}\n", .{self.buffer.items});
}
pub fn relativePathFrom(
self: Routes,
root: enum { root, views, jobs },
sub_path: []const u8,
format: enum { os, posix },
) ![]u8 {
const root_path = switch (root) {
.root => self.root_path,
.views => self.views_path,
.jobs => self.jobs_path,
};
const path = try std.fs.path.relative(self.allocator, root_path, sub_path);
defer self.allocator.free(path);
return switch (format) {
.posix => try self.normalizePosix(path),
.os => try self.allocator.dupe(u8, path),
};
}
fn writeRoutes(self: *Routes, writer: anytype) !void {
var dir = std.fs.openDirAbsolute(self.views_path, .{ .iterate = true }) catch |err| {
switch (err) {
error.FileNotFound => {
std.debug.print(
"[jetzig] Views directory not found, no routes generated: `{s}`\n",
.{self.views_path},
);
return;
},
else => return err,
}
};
defer dir.close();
var walker = try dir.walk(self.allocator);
defer walker.deinit();
while (try walker.next()) |entry| {
@ -114,7 +192,10 @@ pub fn generateRoutes(self: *Self) !void {
if (!std.mem.eql(u8, extension, ".zig")) continue;
const view_routes = try self.generateRoutesForView(views_dir, entry.path);
const realpath = try dir.realpathAlloc(self.allocator, entry.path);
defer self.allocator.free(realpath);
const view_routes = try self.generateRoutesForView(dir, try self.allocator.dupe(u8, realpath));
for (view_routes.static) |view_route| {
try self.static_routes.append(view_route);
@ -128,35 +209,24 @@ pub fn generateRoutes(self: *Self) !void {
std.sort.pdq(Function, self.static_routes.items, {}, Function.lessThanFn);
std.sort.pdq(Function, self.dynamic_routes.items, {}, Function.lessThanFn);
try writer.writeAll(
\\const jetzig = @import("jetzig");
\\
\\pub const routes = [_]jetzig.views.Route{
\\
);
for (self.static_routes.items) |static_route| {
try self.writeRoute(writer, static_route);
}
for (self.dynamic_routes.items) |dynamic_route| {
try self.writeRoute(writer, dynamic_route);
const name = try dynamic_route.fullName(self.allocator);
const name = try dynamic_route.fullName();
defer self.allocator.free(name);
}
std.debug.print("[jetzig] Imported {} route(s)\n", .{self.dynamic_routes.items.len});
try writer.writeAll("};");
// std.debug.print("routes.zig\n{s}\n", .{self.buffer.items});
}
fn writeRoute(self: *Self, writer: std.ArrayList(u8).Writer, route: Function) !void {
const full_name = try route.fullName(self.allocator);
fn writeRoute(self: *Routes, writer: std.ArrayList(u8).Writer, route: Function) !void {
const full_name = try route.fullName();
defer self.allocator.free(full_name);
const uri_path = try route.uriPath(self.allocator);
const uri_path = try route.uriPath();
defer self.allocator.free(uri_path);
const output_template =
@ -164,7 +234,7 @@ fn writeRoute(self: *Self, writer: std.ArrayList(u8).Writer, route: Function) !v
\\ .name = "{0s}",
\\ .action = .{1s},
\\ .view_name = "{2s}",
\\ .view = jetzig.views.Route.ViewType{{ .{3s} = .{{ .{1s} = @import("{7s}").{1s} }} }},
\\ .view = jetzig.Route.ViewType{{ .{3s} = .{{ .{1s} = @import("{7s}").{1s} }} }},
\\ .static = {4s},
\\ .uri_path = "{5s}",
\\ .template = "{6s}",
@ -174,11 +244,10 @@ fn writeRoute(self: *Self, writer: std.ArrayList(u8).Writer, route: Function) !v
\\
;
const module_path = try self.allocator.dupe(u8, route.path);
const module_path = try self.relativePathFrom(.root, route.path, .posix);
defer self.allocator.free(module_path);
const view_name = try self.allocator.dupe(u8, route.view_name);
std.mem.replaceScalar(u8, view_name, '\\', '/');
const view_name = try route.viewName();
defer self.allocator.free(view_name);
const template = try std.mem.concat(self.allocator, u8, &[_][]const u8{ view_name, "/", route.name });
@ -206,9 +275,9 @@ const RouteSet = struct {
static: []Function,
};
fn generateRoutesForView(self: *Self, views_dir: std.fs.Dir, path: []const u8) !RouteSet {
const stat = try views_dir.statFile(path);
const source = try views_dir.readFileAllocOptions(self.allocator, path, stat.size, null, @alignOf(u8), 0);
fn generateRoutesForView(self: *Routes, dir: std.fs.Dir, path: []const u8) !RouteSet {
const stat = try dir.statFile(path);
const source = try dir.readFileAllocOptions(self.allocator, path, stat.size, null, @alignOf(u8), 0);
defer self.allocator.free(source);
self.ast = try std.zig.Ast.parse(self.allocator, source, .zig);
@ -271,7 +340,7 @@ fn generateRoutesForView(self: *Self, views_dir: std.fs.Dir, path: []const u8) !
}
// Parse the `pub const static_params` definition and into a `jetzig.data.Value`.
fn parseStaticParamsDecl(self: *Self, decl: std.zig.Ast.full.VarDecl, params: *jetzig.data.Value) !void {
fn parseStaticParamsDecl(self: *Routes, decl: std.zig.Ast.full.VarDecl, params: *jetzig.data.Value) !void {
const init_node = self.ast.nodes.items(.tag)[decl.ast.init_node];
switch (init_node) {
.struct_init_dot_two, .struct_init_dot_two_comma => {
@ -282,7 +351,7 @@ fn parseStaticParamsDecl(self: *Self, decl: std.zig.Ast.full.VarDecl, params: *j
}
// Recursively parse a struct into a jetzig.data.Value so it can be serialized as JSON and stored
// in `routes.zig` - used for static param comparison at runtime.
fn parseStruct(self: *Self, node: std.zig.Ast.Node.Index, params: *jetzig.data.Value) anyerror!void {
fn parseStruct(self: *Routes, node: std.zig.Ast.Node.Index, params: *jetzig.data.Value) anyerror!void {
var struct_buf: [2]std.zig.Ast.Node.Index = undefined;
const maybe_struct_init = self.ast.fullStructInit(&struct_buf, node);
@ -297,7 +366,7 @@ fn parseStruct(self: *Self, node: std.zig.Ast.Node.Index, params: *jetzig.data.V
}
// Array of param sets for a route, e.g. `.{ .{ .foo = "bar" } }
fn parseArray(self: *Self, node: std.zig.Ast.Node.Index, params: *jetzig.data.Value) anyerror!void {
fn parseArray(self: *Routes, node: std.zig.Ast.Node.Index, params: *jetzig.data.Value) anyerror!void {
var array_buf: [2]std.zig.Ast.Node.Index = undefined;
const maybe_array = self.ast.fullArrayInit(&array_buf, node);
@ -349,7 +418,7 @@ fn parseArray(self: *Self, node: std.zig.Ast.Node.Index, params: *jetzig.data.Va
}
// Parse the value of a param field (recursively when field is a struct/array)
fn parseField(self: *Self, node: std.zig.Ast.Node.Index, params: *jetzig.data.Value) anyerror!void {
fn parseField(self: *Routes, node: std.zig.Ast.Node.Index, params: *jetzig.data.Value) anyerror!void {
const tag = self.ast.nodes.items(.tag)[node];
switch (tag) {
// Route params, e.g. `.index = .{ ... }`
@ -397,7 +466,7 @@ fn parseNumber(value: []const u8, data: *jetzig.data.Data) !*jetzig.data.Value {
}
}
fn isStaticParamsDecl(self: *Self, decl: std.zig.Ast.full.VarDecl) bool {
fn isStaticParamsDecl(self: *Routes, decl: std.zig.Ast.full.VarDecl) bool {
if (decl.visib_token) |token_index| {
const visibility = self.ast.tokenSlice(token_index);
const mutability = self.ast.tokenSlice(decl.ast.mut_token);
@ -411,7 +480,7 @@ fn isStaticParamsDecl(self: *Self, decl: std.zig.Ast.full.VarDecl) bool {
}
fn parseFunction(
self: *Self,
self: *Routes,
index: usize,
path: []const u8,
source: []const u8,
@ -442,7 +511,8 @@ fn parseFunction(
return .{
.name = function_name,
.view_name = try self.allocator.dupe(u8, view_name),
.path = try std.fs.path.join(self.allocator, &[_][]const u8{ "src", "app", "views", path }),
.routes = self,
.path = path,
.args = try self.allocator.dupe(Arg, args.items),
.source = try self.allocator.dupe(u8, source),
.params = std.ArrayList([]const u8).init(self.allocator),
@ -452,7 +522,7 @@ fn parseFunction(
return null;
}
fn parseTypeExpr(self: *Self, node: std.zig.Ast.Node) ![]const u8 {
fn parseTypeExpr(self: *Routes, node: std.zig.Ast.Node) ![]const u8 {
switch (node.tag) {
// Currently all expected params are pointers, keeping this here in case that changes in future:
.identifier => {},
@ -487,3 +557,81 @@ fn isActionFunctionName(name: []const u8) bool {
return false;
}
inline fn chompExtension(path: []const u8) []const u8 {
return path[0 .. path.len - std.fs.path.extension(path).len];
}
fn zigEscape(self: Routes, input: []const u8) ![]const u8 {
var buf = std.ArrayList(u8).init(self.allocator);
const writer = buf.writer();
try std.zig.stringEscape(input, "", .{}, writer);
return try buf.toOwnedSlice();
}
fn normalizePosix(self: Routes, path: []const u8) ![]u8 {
var buf = std.ArrayList([]const u8).init(self.allocator);
defer buf.deinit();
var it = std.mem.splitSequence(u8, path, std.fs.path.sep_str);
while (it.next()) |segment| try buf.append(segment);
return try std.mem.join(self.allocator, std.fs.path.sep_str_posix, buf.items);
}
//
// Generate Jobs
//
fn writeJobs(self: Routes, writer: anytype) !void {
var dir = std.fs.openDirAbsolute(self.jobs_path, .{ .iterate = true }) catch |err| {
switch (err) {
error.FileNotFound => {
std.debug.print(
"[jetzig] Jobs directory not found, no jobs generated: `{s}`\n",
.{self.jobs_path},
);
return;
},
else => return err,
}
};
defer dir.close();
var count: usize = 0;
var walker = try dir.walk(self.allocator);
while (try walker.next()) |entry| {
if (!std.mem.eql(u8, std.fs.path.extension(entry.path), ".zig")) continue;
const realpath = try dir.realpathAlloc(self.allocator, entry.path);
defer self.allocator.free(realpath);
const root_relative_path = try self.relativePathFrom(.root, realpath, .posix);
defer self.allocator.free(root_relative_path);
const jobs_relative_path = try self.relativePathFrom(.jobs, realpath, .posix);
defer self.allocator.free(jobs_relative_path);
const module_path = try self.zigEscape(root_relative_path);
defer self.allocator.free(module_path);
const name_path = try self.zigEscape(jobs_relative_path);
defer self.allocator.free(name_path);
const name = chompExtension(name_path);
try writer.writeAll(try std.fmt.allocPrint(
self.allocator,
\\ .{{
\\ .name = "{0s}",
\\ .runFn = @import("{1s}").run,
\\ }},
\\
,
.{ name, module_path },
));
count += 1;
}
std.debug.print("[jetzig] Imported {} job(s)\n", .{count});
}

25
src/jetzig.zig
View File

@ -2,6 +2,7 @@ const std = @import("std");
pub const zmpl = @import("zmpl").zmpl;
pub const zmd = @import("zmd").zmd;
pub const jetkv = @import("jetkv");
pub const http = @import("jetzig/http.zig");
pub const loggers = @import("jetzig/loggers.zig");
@ -12,6 +13,7 @@ pub const middleware = @import("jetzig/middleware.zig");
pub const util = @import("jetzig/util.zig");
pub const types = @import("jetzig/types.zig");
pub const markdown = @import("jetzig/markdown.zig");
pub const jobs = @import("jetzig/jobs.zig");
/// The primary interface for a Jetzig application. Create an `App` in your application's
/// `src/main.zig` and call `start` to launch the application.
@ -37,8 +39,23 @@ pub const Data = data.Data;
/// generate a `View`.
pub const View = views.View;
/// A route definition. Generated at build type by `Routes.zig`.
pub const Route = views.Route;
const root = @import("root");
/// An asynchronous job that runs outside of the request/response flow. Create via `Request.job`
/// and set params with `Job.put`, then call `Job.schedule()` to add to the
/// job queue.
pub const Job = jobs.Job;
/// A container for a job definition, includes the job name and run function.
pub const JobDefinition = jobs.Job.JobDefinition;
/// A generic logger type. Provides all standard log levels as functions (`INFO`, `WARN`,
/// `ERROR`, etc.). Note that all log functions are CAPITALIZED.
pub const Logger = loggers.Logger;
/// Global configuration. Override these values by defining in `src/main.zig` with:
/// ```zig
/// pub const jetzig_options = struct {
@ -70,6 +87,14 @@ pub const config = struct {
/// A struct of fragments to use when rendering Markdown templates.
pub const markdown_fragments = zmd.html.DefaultFragments;
/// The number of worker threads to spawn on startup for processing Jobs (NOT the number of
/// HTTP server worker threads).
pub const job_worker_threads: usize = 1;
/// Duration before looking for more Jobs when the queue is found to be empty, in
/// milliseconds.
pub const job_worker_sleep_interval_ms: usize = 10;
/// Reconciles a configuration value from user-defined values and defaults provided by Jetzig.
pub fn get(T: type, comptime key: []const u8) T {
const self = @This();

23
src/jetzig/App.zig
View File

@ -5,12 +5,12 @@ const args = @import("args");
const jetzig = @import("../jetzig.zig");
const mime_types = @import("mime_types").mime_types; // Generated at build time.
const Self = @This();
const App = @This();
server_options: jetzig.http.Server.ServerOptions,
allocator: std.mem.Allocator,
pub fn deinit(self: Self) void {
pub fn deinit(self: App) void {
_ = self;
}
@ -21,7 +21,7 @@ const AppOptions = struct {};
/// Starts an application. `routes` should be `@import("routes").routes`, a generated file
/// automatically created at build time. `templates` should be
/// `@import("src/app/views/zmpl.manifest.zig").templates`, created by Zmpl at compile time.
pub fn start(self: Self, routes_module: type, options: AppOptions) !void {
pub fn start(self: App, routes_module: type, options: AppOptions) !void {
_ = options; // See `AppOptions`
var mime_map = jetzig.http.mime.MimeMap.init(self.allocator);
@ -57,6 +57,8 @@ pub fn start(self: Self, routes_module: type, options: AppOptions) !void {
self.allocator.destroy(route);
};
var jet_kv = jetzig.jetkv.JetKV.init(self.allocator, .{});
if (self.server_options.detach) {
const argv = try std.process.argsAlloc(self.allocator);
defer std.process.argsFree(self.allocator, argv);
@ -76,10 +78,25 @@ pub fn start(self: Self, routes_module: type, options: AppOptions) !void {
self.allocator,
self.server_options,
routes.items,
&routes_module.jobs,
&mime_map,
&jet_kv,
);
defer server.deinit();
var worker_pool = jetzig.jobs.Pool.init(
self.allocator,
&jet_kv,
&routes_module.jobs,
server.logger, // TODO: Optional separate log streams for workers
);
defer worker_pool.deinit();
try worker_pool.work(
jetzig.config.get(usize, "job_worker_threads"),
jetzig.config.get(usize, "job_worker_sleep_interval_ms"),
);
server.listen() catch |err| {
switch (err) {
error.AddressInUse => {

56
src/jetzig/http/Request.zig
View File

@ -275,9 +275,63 @@ fn parseQuery(self: *Request) !*jetzig.data.Value {
return self.query_body.?.data.value.?;
}
/// Put a String or Array into the key-value store.
/// `T` can be either `jetzig.KVString` or `jetzig.KVArray`
pub fn kvPut(
self: *Request,
comptime value_type: jetzig.jetkv.value_types,
key: jetzig.jetkv.types.String,
value: jetzig.jetkv.ValueType(value_type),
) !void {
try self.server.jet_kv.put(value_type, key, value);
}
/// Get a String or Array from the key-value store.
/// `T` can be either `jetzig.KVString` or `jetzig.KVArray`
pub fn kvGet(
self: *Request,
comptime value_type: jetzig.jetkv.value_types,
key: jetzig.jetkv.types.String,
) ?jetzig.jetkv.ValueType(value_type) {
return self.server.jet_kv.get(value_type, key);
}
/// Pop a String from an Array in the key-value store.
pub fn kvPop(self: *Request, key: jetzig.jetkv.types.String) ?jetzig.jetkv.types.String {
return self.server.jet_kv.pop(key);
}
/// Return a new Array suitable for use in the KV store.
pub fn kvArray(self: Request) jetzig.jetkv.types.Array {
return jetzig.jetkv.types.Array.init(self.allocator);
}
/// Creates a new Job. Receives a job name which must resolve to `src/app/jobs/<name>.zig`
/// Call `Job.put(...)` to set job params.
/// Call `Job.background()` to run the job outside of the request/response flow.
/// e.g.:
/// ```
/// pub fn post(request: *jetzig.Request, data: *jetzig.Data) !jetzig.View {
/// var job = try request.job("foo"); // Will invoke `process()` in `src/app/jobs/foo.zig`
/// try job.put("foo", data.string("bar"));
/// try job.background(); // Job added to queue and processed by job worker.
/// return request.render(.ok);
/// }
/// ```
pub fn job(self: *Request, job_name: []const u8) !*jetzig.Job {
const background_job = try self.allocator.create(jetzig.Job);
background_job.* = jetzig.Job.init(
self.allocator,
self.server.jet_kv,
self.server.logger,
self.server.job_definitions,
job_name,
);
return background_job;
}
fn extensionFormat(self: *Request) ?jetzig.http.Request.Format {
const extension = self.path.extension orelse return null;
if (std.mem.eql(u8, extension, ".html")) {
return .HTML;
} else if (std.mem.eql(u8, extension, ".json")) {

42
src/jetzig/http/Server.zig
View File

@ -16,34 +16,40 @@ allocator: std.mem.Allocator,
logger: jetzig.loggers.Logger,
options: ServerOptions,
routes: []*jetzig.views.Route,
job_definitions: []const jetzig.JobDefinition,
mime_map: *jetzig.http.mime.MimeMap,
std_net_server: std.net.Server = undefined,
initialized: bool = false,
jet_kv: *jetzig.jetkv.JetKV,
const Self = @This();
const Server = @This();
pub fn init(
allocator: std.mem.Allocator,
options: ServerOptions,
routes: []*jetzig.views.Route,
job_definitions: []const jetzig.JobDefinition,
mime_map: *jetzig.http.mime.MimeMap,
) Self {
jet_kv: *jetzig.jetkv.JetKV,
) Server {
return .{
.allocator = allocator,
.logger = options.logger,
.options = options,
.routes = routes,
.job_definitions = job_definitions,
.mime_map = mime_map,
.jet_kv = jet_kv,
};
}
pub fn deinit(self: *Self) void {
pub fn deinit(self: *Server) void {
if (self.initialized) self.std_net_server.deinit();
self.allocator.free(self.options.secret);
self.allocator.free(self.options.bind);
}
pub fn listen(self: *Self) !void {
pub fn listen(self: *Server) !void {
const address = try std.net.Address.parseIp(self.options.bind, self.options.port);
self.std_net_server = try address.listen(.{ .reuse_port = true });
@ -57,7 +63,7 @@ pub fn listen(self: *Self) !void {
try self.processRequests();
}
fn processRequests(self: *Self) !void {
fn processRequests(self: *Server) !void {
// TODO: Keepalive
while (true) {
var arena = std.heap.ArenaAllocator.init(self.allocator);
@ -82,7 +88,7 @@ fn processRequests(self: *Self) !void {
}
}
fn processNextRequest(self: *Self, allocator: std.mem.Allocator, std_http_server: *std.http.Server) !void {
fn processNextRequest(self: *Server, allocator: std.mem.Allocator, std_http_server: *std.http.Server) !void {
const start_time = std.time.nanoTimestamp();
const std_http_request = try std_http_server.receiveHead();
@ -108,7 +114,7 @@ fn processNextRequest(self: *Self, allocator: std.mem.Allocator, std_http_server
try self.logger.logRequest(&request);
}
fn renderResponse(self: *Self, request: *jetzig.http.Request) !void {
fn renderResponse(self: *Server, request: *jetzig.http.Request) !void {
const static_resource = self.matchStaticResource(request) catch |err| {
if (isUnhandledError(err)) return err;
@ -139,7 +145,7 @@ fn renderStatic(resource: StaticResource, request: *jetzig.http.Request) !void {
}
fn renderHTML(
self: *Self,
self: *Server,
request: *jetzig.http.Request,
route: ?*jetzig.views.Route,
) !void {
@ -169,7 +175,7 @@ fn renderHTML(
}
fn renderJSON(
self: *Self,
self: *Server,
request: *jetzig.http.Request,
route: ?*jetzig.views.Route,
) !void {
@ -191,7 +197,7 @@ fn renderJSON(
}
fn renderMarkdown(
self: *Self,
self: *Server,
request: *jetzig.http.Request,
maybe_route: ?*jetzig.views.Route,
) !?RenderedView {
@ -246,7 +252,7 @@ fn renderMarkdown(
pub const RenderedView = struct { view: jetzig.views.View, content: []const u8 };
fn renderView(
self: *Self,
self: *Server,
route: *jetzig.views.Route,
request: *jetzig.http.Request,
template: ?zmpl.Template,
@ -288,7 +294,7 @@ fn renderView(
}
fn renderTemplateWithLayout(
self: *Self,
self: *Server,
request: *jetzig.http.Request,
template: zmpl.Template,
view: jetzig.views.View,
@ -347,7 +353,7 @@ fn isBadHttpError(err: anyerror) bool {
};
}
fn renderInternalServerError(self: *Self, request: *jetzig.http.Request, err: anyerror) !RenderedView {
fn renderInternalServerError(self: *Server, request: *jetzig.http.Request, err: anyerror) !RenderedView {
request.response_data.reset();
try self.logger.ERROR("Encountered Error: {s}", .{@errorName(err)});
@ -386,7 +392,7 @@ fn renderBadRequest(request: *jetzig.http.Request) !RenderedView {
}
fn logStackTrace(
self: *Self,
self: *Server,
stack: *std.builtin.StackTrace,
request: *jetzig.http.Request,
) !void {
@ -398,7 +404,7 @@ fn logStackTrace(
try self.logger.ERROR("{s}\n", .{buf.items});
}
fn matchRoute(self: *Self, request: *jetzig.http.Request, static: bool) !?*jetzig.views.Route {
fn matchRoute(self: *Server, request: *jetzig.http.Request, static: bool) !?*jetzig.views.Route {
for (self.routes) |route| {
// .index routes always take precedence.
if (route.static == static and route.action == .index and try request.match(route.*)) return route;
@ -413,7 +419,7 @@ fn matchRoute(self: *Self, request: *jetzig.http.Request, static: bool) !?*jetzi
const StaticResource = struct { content: []const u8, mime_type: []const u8 = "application/octet-stream" };
fn matchStaticResource(self: *Self, request: *jetzig.http.Request) !?StaticResource {
fn matchStaticResource(self: *Server, request: *jetzig.http.Request) !?StaticResource {
// TODO: Map public and static routes at launch to avoid accessing the file system when
// matching any route - currently every request causes file system traversal.
const public_resource = try self.matchPublicContent(request);
@ -431,7 +437,7 @@ fn matchStaticResource(self: *Self, request: *jetzig.http.Request) !?StaticResou
return null;
}
fn matchPublicContent(self: *Self, request: *jetzig.http.Request) !?StaticResource {
fn matchPublicContent(self: *Server, request: *jetzig.http.Request) !?StaticResource {
if (request.path.file_path.len <= 1) return null;
if (request.method != .GET) return null;
@ -470,7 +476,7 @@ fn matchPublicContent(self: *Self, request: *jetzig.http.Request) !?StaticResour
return null;
}
fn matchStaticContent(self: *Self, request: *jetzig.http.Request) !?[]const u8 {
fn matchStaticContent(self: *Server, request: *jetzig.http.Request) !?[]const u8 {
var static_dir = std.fs.cwd().openDir("static", .{}) catch |err| {
switch (err) {
error.FileNotFound => return null,

4
src/jetzig/jobs.zig Normal file
View File

@ -0,0 +1,4 @@
pub const Job = @import("jobs/Job.zig");
pub const JobDefinition = Job.JobDefinition;
pub const Pool = @import("jobs/Pool.zig");
pub const Worker = @import("jobs/Worker.zig");

78
src/jetzig/jobs/Job.zig Normal file
View File

@ -0,0 +1,78 @@
const std = @import("std");
const jetzig = @import("../../jetzig.zig");
/// Job name and run function, used when generating an array of job definitions at build time.
pub const JobDefinition = struct {
name: []const u8,
runFn: *const fn (std.mem.Allocator, *jetzig.data.Value, jetzig.loggers.Logger) anyerror!void,
};
allocator: std.mem.Allocator,
jet_kv: *jetzig.jetkv.JetKV,
logger: jetzig.loggers.Logger,
name: []const u8,
definition: ?JobDefinition,
data: ?*jetzig.data.Data = null,
_params: ?*jetzig.data.Value = null,
const Job = @This();
/// Initialize a new Job
pub fn init(
allocator: std.mem.Allocator,
jet_kv: *jetzig.jetkv.JetKV,
logger: jetzig.loggers.Logger,
jobs: []const JobDefinition,
name: []const u8,
) Job {
var definition: ?JobDefinition = null;
for (jobs) |job_definition| {
if (std.mem.eql(u8, job_definition.name, name)) {
definition = job_definition;
break;
}
}
return .{
.allocator = allocator,
.jet_kv = jet_kv,
.logger = logger,
.name = name,
.definition = definition,
};
}
/// Deinitialize the Job and frees memory
pub fn deinit(self: *Job) void {
if (self.data) |data| {
data.deinit();
self.allocator.destroy(data);
}
}
/// Add a parameter to the Job
pub fn put(self: *Job, key: []const u8, value: *jetzig.data.Value) !void {
var job_params = try self.params();
try job_params.put(key, value);
}
/// Add a Job to the queue
pub fn schedule(self: *Job) !void {
_ = try self.params();
const json = try self.data.?.toJson();
try self.jet_kv.prepend("__jetzig_jobs", json);
try self.logger.INFO("Scheduled job: {s}", .{self.name});
}
fn params(self: *Job) !*jetzig.data.Value {
if (self.data == null) {
self.data = try self.allocator.create(jetzig.data.Data);
self.data.?.* = jetzig.data.Data.init(self.allocator);
self._params = try self.data.?.object();
try self._params.?.put("__jetzig_job_name", self.data.?.string(self.name));
}
return self._params.?;
}
// TODO: Tests :)

55
src/jetzig/jobs/Pool.zig Normal file
View File

@ -0,0 +1,55 @@
const std = @import("std");
const jetzig = @import("../../jetzig.zig");
const Pool = @This();
allocator: std.mem.Allocator,
jet_kv: *jetzig.jetkv.JetKV,
job_definitions: []const jetzig.jobs.JobDefinition,
logger: jetzig.loggers.Logger,
pool: std.Thread.Pool = undefined,
workers: std.ArrayList(*jetzig.jobs.Worker),
/// Initialize a new worker thread pool.
pub fn init(
allocator: std.mem.Allocator,
jet_kv: *jetzig.jetkv.JetKV,
job_definitions: []const jetzig.jobs.JobDefinition,
logger: jetzig.loggers.Logger,
) Pool {
return .{
.allocator = allocator,
.jet_kv = jet_kv,
.job_definitions = job_definitions,
.logger = logger,
.workers = std.ArrayList(*jetzig.jobs.Worker).init(allocator),
};
}
/// Free pool resources and destroy workers.
pub fn deinit(self: *Pool) void {
self.pool.deinit();
for (self.workers.items) |worker| self.allocator.destroy(worker);
self.workers.deinit();
}
/// Spawn a given number of threads and start processing jobs, sleep for a given interval (ms)
/// when no jobs are in the queue. Each worker operates its own work loop.
pub fn work(self: *Pool, threads: usize, interval: usize) !void {
try self.pool.init(.{ .allocator = self.allocator });
for (0..threads) |index| {
const worker = try self.allocator.create(jetzig.jobs.Worker);
worker.* = jetzig.jobs.Worker.init(
self.allocator,
self.logger,
index,
self.jet_kv,
self.job_definitions,
interval,
);
try self.workers.append(worker);
try self.pool.spawn(jetzig.jobs.Worker.work, .{worker});
}
}

123
src/jetzig/jobs/Worker.zig Normal file
View File

@ -0,0 +1,123 @@
const std = @import("std");
const jetzig = @import("../../jetzig.zig");
const Worker = @This();
allocator: std.mem.Allocator,
logger: jetzig.loggers.Logger,
id: usize,
jet_kv: *jetzig.jetkv.JetKV,
job_definitions: []const jetzig.jobs.JobDefinition,
interval: usize,
pub fn init(
allocator: std.mem.Allocator,
logger: jetzig.loggers.Logger,
id: usize,
jet_kv: *jetzig.jetkv.JetKV,
job_definitions: []const jetzig.jobs.JobDefinition,
interval: usize,
) Worker {
return .{
.allocator = allocator,
.logger = logger,
.id = id,
.jet_kv = jet_kv,
.job_definitions = job_definitions,
.interval = interval * 1000 * 1000, // millisecond => nanosecond
};
}
/// Begin working through jobs in the queue.
pub fn work(self: *const Worker) void {
self.log(.INFO, "[worker-{}] Job worker started.", .{self.id});
while (true) {
if (self.jet_kv.pop("__jetzig_jobs")) |json| {
defer self.allocator.free(json);
if (self.matchJob(json)) |job_definition| {
self.processJob(job_definition, json);
}
} else {
std.time.sleep(self.interval);
}
}
self.log(.INFO, "[worker-{}] Job worker exited.", .{self.id});
}
// Do a minimal parse of JSON job data to identify job name, then match on known job definitions.
fn matchJob(self: Worker, json: []const u8) ?jetzig.jobs.JobDefinition {
const parsed_json = std.json.parseFromSlice(
struct { __jetzig_job_name: []const u8 },
self.allocator,
json,
.{ .ignore_unknown_fields = true },
) catch |err| {
self.log(
.ERROR,
"[worker-{}] Error parsing JSON from job queue: {s}",
.{ self.id, @errorName(err) },
);
return null;
};
const job_name = parsed_json.value.__jetzig_job_name;
// TODO: Hashmap
for (self.job_definitions) |job_definition| {
if (std.mem.eql(u8, job_definition.name, job_name)) {
parsed_json.deinit();
return job_definition;
}
} else {
self.log(.WARN, "[worker-{}] Tried to process unknown job: {s}", .{ self.id, job_name });
return null;
}
}
// Fully parse JSON job data and invoke the defined job's run function, passing the parsed params
// as a `*jetzig.data.Value`.
fn processJob(self: Worker, job_definition: jetzig.JobDefinition, json: []const u8) void {
var data = jetzig.data.Data.init(self.allocator);
defer data.deinit();
data.fromJson(json) catch |err| {
self.log(
.INFO,
"[worker-{}] Error parsing JSON for job `{s}`: {s}",
.{ self.id, job_definition.name, @errorName(err) },
);
};
var arena = std.heap.ArenaAllocator.init(self.allocator);
defer arena.deinit();
if (data.value) |params| {
job_definition.runFn(arena.allocator(), params, self.logger) catch |err| {
self.log(
.ERROR,
"[worker-{}] Encountered error processing job `{s}`: {s}",
.{ self.id, job_definition.name, @errorName(err) },
);
return;
};
self.log(.INFO, "[worker-{}] Job completed: {s}", .{ self.id, job_definition.name });
} else {
self.log(.ERROR, "Error in job params definition for job: {s}", .{job_definition.name});
}
}
// Log with error handling and fallback. Prefix with worker ID.
fn log(
self: Worker,
comptime level: jetzig.loggers.LogLevel,
comptime message: []const u8,
args: anytype,
) void {
self.logger.log(level, message, args) catch |err| {
// XXX: In (daemonized) deployment stderr will not be available, find a better solution.
// Note that this only occurs if logging itself fails.
std.debug.print("[worker-{}] Logger encountered error: {s}\n", .{ self.id, @errorName(err) });
};
}

11
src/jetzig/loggers.zig
View File

@ -61,4 +61,15 @@ pub const Logger = union(enum) {
inline else => |*logger| try logger.logRequest(request),
}
}
pub fn log(
self: *const Logger,
comptime level: LogLevel,
comptime message: []const u8,
args: anytype,
) !void {
switch (self.*) {
inline else => |*logger| try logger.log(level, message, args),
}
}
};

7
src/jetzig/loggers/DevelopmentLogger.zig
View File

@ -13,6 +13,7 @@ stderr: std.fs.File,
stdout_colorized: bool,
stderr_colorized: bool,
level: LogLevel,
mutex: std.Thread.Mutex,
/// Initialize a new Development Logger.
pub fn init(
@ -28,12 +29,13 @@ pub fn init(
.stderr = stderr,
.stdout_colorized = stdout.isTty(),
.stderr_colorized = stderr.isTty(),
.mutex = std.Thread.Mutex{},
};
}
/// Generic log function, receives log level, message (format string), and args for format string.
pub fn log(
self: DevelopmentLogger,
self: *const DevelopmentLogger,
comptime level: LogLevel,
comptime message: []const u8,
args: anytype,
@ -58,6 +60,9 @@ pub fn log(
const writer = file.writer();
const level_formatted = if (colorized) colorizedLogLevel(level) else @tagName(level);
@constCast(self).mutex.lock();
defer @constCast(self).mutex.unlock();
try writer.print("{s: >5} [{s}] {s}\n", .{ level_formatted, iso8601, output });
if (!file.isTty()) try file.sync();

12
src/jetzig/loggers/JsonLogger.zig
View File

@ -24,6 +24,7 @@ allocator: std.mem.Allocator,
stdout: std.fs.File,
stderr: std.fs.File,
level: LogLevel,
mutex: std.Thread.Mutex,
/// Initialize a new JSON Logger.
pub fn init(
@ -37,12 +38,13 @@ pub fn init(
.level = level,
.stdout = stdout,
.stderr = stderr,
.mutex = std.Thread.Mutex{},
};
}
/// Generic log function, receives log level, message (format string), and args for format string.
pub fn log(
self: JsonLogger,
self: *const JsonLogger,
comptime level: LogLevel,
comptime message: []const u8,
args: anytype,
@ -63,6 +65,9 @@ pub fn log(
const json = try std.json.stringifyAlloc(self.allocator, log_message, .{ .whitespace = .minified });
defer self.allocator.free(json);
@constCast(self).mutex.lock();
defer @constCast(self).mutex.unlock();
try writer.writeAll(json);
try writer.writeByte('\n');
@ -70,7 +75,7 @@ pub fn log(
}
/// Log a one-liner including response status code, path, method, duration, etc.
pub fn logRequest(self: JsonLogger, request: *const jetzig.http.Request) !void {
pub fn logRequest(self: *const JsonLogger, request: *const jetzig.http.Request) !void {
const level: LogLevel = .INFO;
const duration = jetzig.util.duration(request.start_time);
@ -100,6 +105,9 @@ pub fn logRequest(self: JsonLogger, request: *const jetzig.http.Request) !void {
const file = self.getFile(level);
const writer = file.writer();
@constCast(self).mutex.lock();
defer @constCast(self).mutex.unlock();
try writer.writeAll(json);
try writer.writeByte('\n');

2
src/tests.zig
View File

@ -3,5 +3,5 @@ test {
_ = @import("jetzig/http/Headers.zig");
_ = @import("jetzig/http/Cookies.zig");
_ = @import("jetzig/http/Path.zig");
@import("std").testing.refAllDeclsRecursive(@This());
_ = @import("jetzig/jobs/Job.zig");
}