add BufferedStreamSource

4 files changed, 201 insertions, 13 deletions

diff --git a/src/control.zig b/src/control.zig
index f547a4b..4a29309 100644
--- a/src/control.zig
+++ b/src/control.zig
@@ -6,7 +6,14 @@ const cfg = @import("config.zig");
 const util = @import("util.zig");
 const build_config = @import("build_config");
 
-fn verify_args(expected: u8, got: []const u8) !void {
+fn verify_args(expected: []const u8, got: []const u8) !void {
+    if (!std.mem.eql(u8, expected, got)) {
+        std.debug.print("expected '{s}' but got {s}\n", .{ expected, got });
+        return error.typeMismatch;
+    }
+}
+
+fn verify_args_all(expected: u8, got: []const u8) !void {
     for (got) |typ| {
         if (typ != expected) {
             std.debug.print("expected '{c}' but got '{c}' element (args are {s})\n", .{ expected, typ, got });
@@ -26,19 +33,19 @@ fn set_one(
 
     switch (T) {
         f32 => {
-            try verify_args('f', types);
+            try verify_args_all('f', types);
             dest.* = argv[0].*.f;
         },
         f64 => {
-            try verify_args('d', types);
+            try verify_args_all('d', types);
             dest.* = argv[0].*.d;
         },
         i32 => {
-            try verify_args('d', types);
+            try verify_args_all('d', types);
             dest.* = argv[0].*.i;
         },
         u32 => {
-            try verify_args('d', types);
+            try verify_args_all('d', types);
             const val = argv[0].*.i;
             if (val < 0)
                 return error.signDisallowed;
@@ -66,22 +73,22 @@ fn set_array(
 
     switch (T) {
         f32 => {
-            try verify_args('f', types);
+            try verify_args_all('f', types);
             for (dest, 0..) |*v, i|
                 v.* = argv[i].*.f;
         },
         f64 => {
-            try verify_args('d', types);
+            try verify_args_all('d', types);
             for (dest, 0..) |*v, i|
                 v.* = argv[i].*.d;
         },
         i32 => {
-            try verify_args('i', types);
+            try verify_args_all('i', types);
             for (dest, 0..) |*v, i|
                 v.* = argv[i].*.i;
         },
         u32 => {
-            try verify_args('i', types);
+            try verify_args_all('i', types);
             for (dest, 0..) |*v, i| {
                 const val = argv[i].*.i;
                 if (val < 0)
@@ -136,6 +143,7 @@ pub const ControlServer = struct {
         if (build_config.have_ffmpeg) {
             try self.add_method(&.{ "/source", "*", "video" }, null, handle_texture_video);
             try self.add_method(&.{ "/source", "*", "stream" }, null, handle_texture_stream);
+            try self.add_method(&.{ "/source", "*", "buffered-stream" }, null, handle_texture_buffered_stream);
         }
         if (build_config.have_tsv) {
             try self.add_method(&.{ "/source", "*", "tsv" }, "ss", handle_texture_tsv);
@@ -274,6 +282,22 @@ pub const ControlServer = struct {
             }
         }
 
+        if (std.mem.eql(u8, types, "s")) {
+            const param_stringZ: [*:0]const u8 = @ptrCast(&argv[0].*.s);
+            const param_string = std.mem.span(param_stringZ);
+            if (std.mem.startsWith(u8, param_string, "/source/")) {
+                var param_parts = std.mem.tokenizeScalar(u8, param_string, '/');
+                _ = param_parts.next(); // skip /source
+                const source_name = param_parts.next().?;
+                const param_name = param_parts.next().?;
+
+                const source = self.sources.get(source_name) orelse return error.texNotFound;
+                try source.updateUniform(param_name, value);
+                uniform.setShaderValue(self.cache.shader.*);
+                return true;
+            }
+        }
+
         try switch (value) {
             .FLOAT => |val| set_one(f32, val, argv, types),
             .DOUBLE => |val| set_one(f64, val, argv, types),
@@ -347,7 +371,7 @@ pub const ControlServer = struct {
     }
 
     fn handle_texture_video(self: *ControlServer, path: []const u8, types: []const u8, argv: []const [*c]c.lo_arg) !bool {
-        try verify_args('s', types);
+        try verify_args_all('s', types);
 
         var parts = std.mem.tokenizeScalar(u8, path, '/');
         _ = parts.next();
@@ -368,7 +392,7 @@ pub const ControlServer = struct {
     }
 
     fn handle_texture_stream(self: *ControlServer, path: []const u8, types: []const u8, argv: []const [*c]c.lo_arg) !bool {
-        try verify_args('s', types);
+        try verify_args_all('s', types);
 
         var parts = std.mem.tokenizeScalar(u8, path, '/');
         _ = parts.next();
@@ -389,6 +413,29 @@ pub const ControlServer = struct {
         return true;
     }
 
+    fn handle_texture_buffered_stream(self: *ControlServer, path: []const u8, types: []const u8, argv: []const [*c]c.lo_arg) !bool {
+        try verify_args("sis", types[0..3]);
+        try verify_args_all('s', types[3..]);
+
+        var parts = std.mem.tokenizeScalar(u8, path, '/');
+        _ = parts.next();
+        const name = parts.next() orelse unreachable;
+        if (self.sources.contains(name)) return error.textureNameTaken;
+
+        const texture_typeZ: [*:0]const u8 = @ptrCast(&argv[0].*.s);
+        const texture_type = std.meta.stringToEnum(gl.Texture.Type, std.mem.span(texture_typeZ)) orelse return error.invalidType;
+        const depth: i32 = argv[1].*.i;
+        const filenameZ: [*:0]const u8 = @ptrCast(&argv[2].*.s);
+        const formatZ: ?[*:0]const u8 = if (argv.len > 3) @ptrCast(&argv[3].*.s) else null;
+        const args = if (argv.len > 4) argv[4..] else &.{};
+
+        const ffmpeg = @import("ffmpeg.zig");
+        const source = try ffmpeg.BufferedStreamSource.init(self.cache.allocator, self.constants, texture_type, depth, filenameZ, formatZ, args);
+        try self.add_source(name, source);
+
+        return true;
+    }
+
     fn handle_texture_tsv(self: *ControlServer, path: []const u8, types: []const u8, argv: []const [*c]c.lo_arg) !bool {
         _ = types;
 
diff --git a/src/ffmpeg.zig b/src/ffmpeg.zig
index fdf9068..6846eb1 100644
--- a/src/ffmpeg.zig
+++ b/src/ffmpeg.zig
@@ -40,7 +40,6 @@ fn get_format_context(
     var options: ?*c.AVDictionary = null;
     var i: usize = 0;
     std.debug.assert(format_options.len % 2 == 0);
-    std.debug.print("len {}\n", .{format_options.len});
     while (i < format_options.len) : (i += 2) {
         try check(c.av_dict_set(
             &options,
@@ -421,3 +420,138 @@ pub const StreamSource = struct {
         allocator.destroy(self);
     }
 };
+
+pub const BufferedStreamSource = struct {
+    source: src.Source,
+    flags: src.StreamFlags,
+
+    decoder: *Decoder,
+    format: *c.AVFormatContext,
+    stream: i32,
+    depth: i32,
+    packet: *c.AVPacket,
+
+    thread: *gl.Thread,
+
+    pub fn init(
+        allocator: std.mem.Allocator,
+        constants: *const gl.Constants,
+        texture_type: gl.Texture.Type,
+        depth: i32,
+        filename: [*:0]const u8,
+        format_name: ?[*:0]const u8,
+        format_options: []const [*c]c.lo_arg,
+    ) !*src.Source {
+        const self = try allocator.create(BufferedStreamSource);
+        errdefer allocator.destroy(self);
+
+        var format = try get_format_context(filename, format_name, format_options);
+        errdefer c.avformat_close_input(@ptrCast(&format));
+        format.flags |= c.AVFMT_FLAG_NONBLOCK;
+
+        try check(c.avformat_find_stream_info(format, null));
+
+        const video_stream = for (format.streams, 0..format.nb_streams) |c_stream, _| {
+            const stream = @as(*c.AVStream, c_stream orelse unreachable);
+            if (stream.codecpar.*.codec_type == c.AVMEDIA_TYPE_VIDEO) break stream;
+        } else unreachable;
+
+        const codec_par = video_stream.codecpar.*;
+
+        var packet = c.av_packet_alloc();
+        errdefer c.av_packet_free(&packet);
+
+        const texture = switch (texture_type) {
+            .TEXTURE_3D, .TEXTURE_2D_ARRAY => |t| gl.Texture.create(t),
+            else => return error.textureTypeInvalid,
+        };
+        texture.allocate(
+            codec_par.width,
+            codec_par.height,
+            depth,
+            c.GL_RGBA8,
+        );
+
+        const decoder = try AVDecoder.init(allocator, codec_par);
+        errdefer decoder.deinit(allocator);
+
+        self.* = .{
+            .source = .{
+                .texture = texture,
+                .deinit_fn = deinit,
+                .register_fn = register_methods,
+                .unregister_fn = unregister_methods,
+                .update_uniform_fn = update_uniform,
+            },
+            .flags = .{},
+
+            .decoder = decoder,
+            .format = format,
+            .stream = video_stream.index,
+            .depth = depth,
+            .packet = packet,
+
+            .thread = undefined,
+        };
+
+        try self.update();
+        self.thread = try gl.Thread.init(allocator, constants, update_loop, .{self});
+
+        return &self.source;
+    }
+
+    fn register_methods(source: *src.Source, name: []const u8, control: *ctrl.ControlServer) void {
+        const self: *BufferedStreamSource = @fieldParentPtr("source", source);
+        self.flags.register(name, control);
+    }
+    fn unregister_methods(source: *src.Source, name: []const u8, control: *ctrl.ControlServer) void {
+        const self: *BufferedStreamSource = @fieldParentPtr("source", source);
+        self.flags.unregister(name, control);
+    }
+
+    fn update_uniform(source: *src.Source, param: []const u8, value: gl.UniformPointer) !void {
+        const self: *BufferedStreamSource = @fieldParentPtr("source", source);
+
+        if (!std.mem.eql(u8, param, "offset")) return error.unknownSourceParam;
+        const frame = @rem(self.decoder.num_frames + self.depth - 1, self.depth);
+        try switch (value) {
+            .FLOAT => |val| val.* = @as(f32, @floatFromInt(frame)) / @as(f32, @floatFromInt(self.depth)),
+            .DOUBLE => |val| val.* = @as(f64, @floatFromInt(frame)) / @as(f64, @floatFromInt(self.depth)),
+            .INT => |val| val.* = frame,
+            .UNSIGNED_INT => |val| val.* = @intCast(frame),
+            else => error.uniformNotSupported,
+        };
+    }
+
+    fn update_loop(self: *BufferedStreamSource) !void {
+        while (!self.thread.quit) {
+            try self.update();
+            c.glFlush();
+            try std.Thread.yield();
+        }
+    }
+
+    fn update(self: *BufferedStreamSource) !void {
+        try check(c.av_read_frame(self.format, self.packet));
+        defer c.av_packet_unref(self.packet);
+
+        if (self.packet.*.stream_index != self.stream) return;
+
+        _ = try self.decoder.process_packet_fn(
+            self.decoder,
+            self.packet,
+            if (self.flags.shouldStep()) &self.source.texture else null,
+        );
+        self.decoder.num_frames = @rem(self.decoder.num_frames, self.depth);
+    }
+
+    fn deinit(source: *src.Source, allocator: std.mem.Allocator) void {
+        const self: *BufferedStreamSource = @fieldParentPtr("source", source);
+
+        self.thread.deinit(allocator);
+        self.decoder.deinit(allocator);
+        c.av_packet_free(@ptrCast(&self.packet));
+        c.avformat_close_input(@ptrCast(&self.format));
+        allocator.destroy(self);
+    }
+};
diff --git a/src/gl.zig b/src/gl.zig
index 87fc9cb..7cdc204 100644
--- a/src/gl.zig
+++ b/src/gl.zig
@@ -386,7 +386,7 @@ const UniformValue = union(UniformType) {
     }
 };
 
-const UniformPointer = union(UniformType) {
+pub const UniformPointer = union(UniformType) {
     FLOAT: *f32,
     FLOAT_VEC2: *[2]f32,
     FLOAT_VEC3: *[3]f32,
diff --git a/src/source.zig b/src/source.zig
index 21e82c8..917e08a 100644
--- a/src/source.zig
+++ b/src/source.zig
@@ -9,6 +9,7 @@ pub const Source = struct {
     deinit_fn: *const fn (self: *Source, allocator: std.mem.Allocator) void,
     register_fn: ?*const fn (self: *Source, name: []const u8, control: *ctrl.ControlServer) void = null,
     unregister_fn: ?*const fn (self: *Source, name: []const u8, control: *ctrl.ControlServer) void = null,
+    update_uniform_fn: ?*const fn (self: *Source, param: []const u8, value: gl.UniformPointer) anyerror!void = null,
 
     pub fn deinit(self: *Source, allocator: std.mem.Allocator) void {
         self.deinit_fn(self, allocator);
@@ -25,6 +26,12 @@ pub const Source = struct {
             unregister_fn(self, name, control);
         }
     }
+
+    pub fn updateUniform(self: *Source, param: []const u8, value: gl.UniformPointer) anyerror!void {
+        if (self.update_uniform_fn) |update_uniform_fn| {
+            try update_uniform_fn(self, param, value);
+        }
+    }
 };
 
 pub const StreamFlags = struct {