Refactor code to use membrane core 0.12

Author: gBillal

c_src/turbojpeg/turbojpeg_native.c

@@ -45,7 +45,7 @@ UNIFEX_TERM get_jpeg_header(UnifexEnv* env, UnifexPayload *payload) {
   enum TJSAMP tjsamp;
   enum TJCS cspace;
   int res, width, height;
-  ERL_NIF_TERM map_out, ret;
+  UNIFEX_TERM ret;
 
   tjh = tjInitDecompress();
   if(!tjh)
@@ -63,45 +63,24 @@ UNIFEX_TERM get_jpeg_header(UnifexEnv* env, UnifexPayload *payload) {
     goto cleanup;
   }
 
-  // unifex does not support maps yes.
-  // See https://github.com/membraneframework/unifex/issues/32
-  if(!enif_make_map_from_arrays(
-    env,
-    (ERL_NIF_TERM []) {
-      enif_make_atom(env, "width"),
-      enif_make_atom(env, "height"),
-      enif_make_atom(env, "format"),
-
-    },
-    (ERL_NIF_TERM []) {
-      enif_make_int(env, width),
-      enif_make_int(env, height),
-      enif_make_atom(env, tjsamp_to_format(tjsamp))
-    },
-    3, &map_out
-  )) {
-    ret = get_jpeg_header_result_error(env, "make_map");
-    goto cleanup;
-  } else {
-    // Generated code does not support maps currently.
-    ret = enif_make_tuple_from_array(env, (ERL_NIF_TERM []) {enif_make_atom(env, "ok"), map_out}, 2);
-    goto cleanup;
-  }
+  jpeg_header header = {(char*)tjsamp_to_format(tjsamp), width, height};
+  ret = get_jpeg_header_result_ok(env, header);
+
 cleanup:
   if(tjh) tjDestroy(tjh);
   return ret;
 }
 
 /**
- * Convert a binary h264 payload into a jpeg encoded payload
+ * Convert a yuv binary payload into a jpeg encoded payload
 */
 UNIFEX_TERM yuv_to_jpeg(UnifexEnv* env, UnifexPayload *payload, int width, int height, int quality, char* format) {
   tjhandle tjh = NULL;
   enum TJSAMP tjsamp;
   unsigned char *jpegBuf = NULL;
   unsigned long jpegSize;
   int res;
-  UnifexPayload *jpegFrame;
+  UnifexPayload *jpegFrame = NULL;
   UNIFEX_TERM ret;
 
   res = format_to_tjsamp(format);
@@ -126,18 +105,19 @@ UNIFEX_TERM yuv_to_jpeg(UnifexEnv* env, UnifexPayload *payload, int width, int h
   if(res < 0) {
     ret = yuv_to_jpeg_result_error(env, tjGetErrorStr2(tjh));
     goto cleanup;
-  } else {
-    jpegFrame = unifex_payload_alloc(env, UNIFEX_PAYLOAD_BINARY, jpegSize);
-    if(!jpegFrame) {
-      ret = yuv_to_jpeg_result_error(env, "payload_alloc");
-    } else {
-      memcpy(jpegFrame->data, jpegBuf, jpegSize);
-      ret = yuv_to_jpeg_result_ok(env, jpegFrame);
-    }
-    goto cleanup;
   }
 
+  jpegFrame = unifex_alloc(sizeof(*jpegFrame));
+  unifex_payload_alloc(env, UNIFEX_PAYLOAD_BINARY, jpegSize, jpegFrame);
+  memcpy(jpegFrame->data, jpegBuf, jpegSize);
+  ret = yuv_to_jpeg_result_ok(env, jpegFrame);
+  
 cleanup:
+  if(jpegFrame != NULL) {
+    unifex_payload_release(jpegFrame);
+    unifex_free(jpegFrame);
+  } 
+    
   if(jpegBuf) tjFree(jpegBuf);
   if(tjh) tjDestroy(tjh);
   return ret;
@@ -151,7 +131,7 @@ UNIFEX_TERM jpeg_to_yuv(UnifexEnv* env, UnifexPayload *payload) {
   enum TJSAMP tjsamp;
   enum TJCS cspace;
   unsigned long yuvBufSize;
-  UnifexPayload *yuvFrame;
+  UnifexPayload *yuvFrame = NULL;
   int res, width, height;
   UNIFEX_TERM ret;
 
@@ -166,17 +146,15 @@ UNIFEX_TERM jpeg_to_yuv(UnifexEnv* env, UnifexPayload *payload) {
     &width, &height, 
     (int*)&tjsamp, (int*)&cspace
   );
+
   if(res < 0) {
     ret = jpeg_to_yuv_result_error(env, tjGetErrorStr2(tjh));
     goto cleanup;
   } 
 
   yuvBufSize = tjBufSizeYUV2(width, 4, height, tjsamp);
-  yuvFrame = unifex_payload_alloc(env, UNIFEX_PAYLOAD_BINARY, yuvBufSize);
-  if(!yuvFrame) {
-    ret = jpeg_to_yuv_result_error(env, "could not allocate frame");
-    goto cleanup;
-  }
+  yuvFrame = unifex_alloc(sizeof(*yuvFrame));
+  unifex_payload_alloc(env, UNIFEX_PAYLOAD_BINARY, yuvBufSize, yuvFrame);
 
   res = tjDecompressToYUV2(
     tjh, 
@@ -190,12 +168,16 @@ UNIFEX_TERM jpeg_to_yuv(UnifexEnv* env, UnifexPayload *payload) {
   if(res < 0) {
     ret = jpeg_to_yuv_result_error(env, tjGetErrorStr2(tjh));
     goto cleanup;
-  } else {
-    ret = jpeg_to_yuv_result_ok(env, yuvFrame);
-    goto cleanup;
   }
+    
+  ret = jpeg_to_yuv_result_ok(env, yuvFrame);
 
 cleanup:
+  if(yuvFrame != NULL) {
+    unifex_payload_release(yuvFrame);
+    unifex_free(yuvFrame);
+  }
+
   if(tjh) tjDestroy(tjh);
   return ret;
 }

c_src/turbojpeg/turbojpeg_native.spec.exs

@@ -1,6 +1,13 @@
 module Turbojpeg.Native
 
-spec yuv_to_jpeg(payload, width::int, height::int, quality::int, format::atom) :: {:ok :: label, payload} | {:error :: label, reason :: atom}
-spec jpeg_to_yuv(payload) :: {:ok :: label, payload} | {:error :: label, reason :: atom}
-spec get_jpeg_header(payload) :: {:ok :: label, data::int} | {:error::label, reason::atom}
-dirty :cpu, yuv_to_jpeg: 5, jpeg_to_yuv: 1, get_jpeg_header: 1
+type jpeg_header :: %Turbojpeg.JpegHeader{
+  format: atom,
+  width: int,
+  height: int
+}
+
+spec yuv_to_jpeg(payload, width::int, height::int, quality::int, format::atom) :: {:ok :: label, payload} | {:error :: label, reason :: string}
+spec jpeg_to_yuv(payload) :: {:ok :: label, payload} | {:error :: label, reason :: string}
+spec get_jpeg_header(payload) :: {:ok :: label, jpeg_header} | {:error::label, reason :: string}
+
+dirty :cpu, yuv_to_jpeg: 5, jpeg_to_yuv: 1, get_jpeg_header: 1

lib/turbojpeg.ex

@@ -1,6 +1,7 @@
 defmodule Turbojpeg do
   @moduledoc File.read!("README.md")
-  alias Turbojpeg.Native
+
+  alias Turbojpeg.{JpegHeader, Native}
 
   @type width :: dimension
   @type height :: dimension
@@ -11,14 +12,7 @@ defmodule Turbojpeg do
           | :I422
           | :I444
           | :GRAY
-  @type error ::
-          {:error, atom()}
-          | {:error, struct}
-  @type jpeg_header :: %{
-          format: format(),
-          width: width(),
-          height: height()
-        }
+  @type error :: {:error, atom()} | {:error, struct}
 
   @doc """
   Converts yuv to jpeg images
@@ -31,9 +25,6 @@ defmodule Turbojpeg do
           {:ok, binary()} | error()
   def yuv_to_jpeg(yuv, width, height, quality, format) do
     Native.yuv_to_jpeg(yuv, width, height, quality, format)
-  rescue
-    error ->
-      {:error, error}
   end
 
   @doc """
@@ -43,32 +34,30 @@ defmodule Turbojpeg do
         iex> {:ok, yuv} = Turbojpeg.jpeg_to_yuv(jpeg)
           {:ok,<<..>>}
   """
-  @spec jpeg_to_yuv(binary()) ::
-          {:ok, binary()} | error()
+  @spec jpeg_to_yuv(binary()) :: {:ok, binary()} | error()
   def jpeg_to_yuv(jpeg) do
     Native.jpeg_to_yuv(jpeg)
-  rescue
-    error ->
-      {:error, error}
   end
 
   @doc """
-  Gets the header from a jpegv
+  Gets the header from a jpeg binary
 
-       iex> {:ok, header} = Turbojpeg.get_jpeg_header(jpeg)
+  ## Examples
+
+      iex> {:ok, header} = Turbojpeg.get_jpeg_header(jpeg)
          {:ok,
-           %{
+           %Turbojpeg.JpegHeader{
               format: :I422,
               width: 192,
               height: 192
             }
          }
+
+      iex> Turbojpeg.get_jpeg_header(<<45, 48, 44, 41, 11>>)
+         {:error, "Not a JPEG file: starts with 0x2d 0x30"}
   """
-  @spec get_jpeg_header(binary()) :: {:ok, jpeg_header} | error()
+  @spec get_jpeg_header(binary()) :: {:ok, JpegHeader.t()} | error()
   def get_jpeg_header(jpeg) do
     Native.get_jpeg_header(jpeg)
-  rescue
-    error ->
-      {:error, error}
   end
 end

lib/turbojpeg/filter.ex

@@ -3,45 +3,49 @@ defmodule Turbojpeg.Filter do
   Membrane filter converting raw video frames to JPEG.
   """
   use Membrane.Filter
+
   alias Membrane.Buffer
+  alias Membrane.RawVideo
+  alias Membrane.RemoteStream
+
+  def_input_pad :input,
+    flow_control: :auto,
+    demand_unit: :buffers,
+    accepted_format: %RawVideo{pixel_format: pix_fmt} when pix_fmt in [:I420, :I422, :I444]
 
-  def_input_pad :input, demand_unit: :buffers, caps: Membrane.Caps.Video.Raw
-  # TODO: implement JPEG caps
-  def_output_pad :output, caps: :any
+  # TODO: implement JPEG stream format
+  def_output_pad :output, accepted_format: RemoteStream
 
   def_options quality: [
-                type: :integer,
                 spec: Turbojpeg.quality(),
-                default: 100,
+                default: 75,
                 description: "Jpeg encoding quality"
               ]
 
   @impl true
-  def handle_demand(:output, size, :buffers, _ctx, state) do
-    {{:ok, demand: {:input, size}}, state}
+  def handle_init(_ctx, options) do
+    {[], Map.from_struct(options)}
   end
 
   @impl true
-  def handle_caps(:input, _caps, _ctx, state) do
-    {:ok, state}
+  def handle_stream_format(:input, _stream_format, _ctx, state) do
+    {[stream_format: {:output, %RemoteStream{type: :bytestream}}], state}
   end
 
   @impl true
-  def handle_process(:input, buffer, ctx, state) do
-    %{caps: caps} = ctx.pads.input
-
-    with {:ok, payload} <-
-           Turbojpeg.yuv_to_jpeg(
-             buffer.payload,
-             caps.width,
-             caps.height,
-             state.quality,
-             caps.format
-           ) do
-      buffer = %Buffer{buffer | payload: payload}
-      {{:ok, buffer: {:output, buffer}}, state}
-    else
-      {:error, _} = error -> {error, state}
+  def handle_process(:input, %Buffer{payload: payload} = buffer, ctx, state) do
+    %{stream_format: stream_format} = ctx.pads.input
+    %{width: width, height: height, pixel_format: pix_fmt} = stream_format
+
+    case Turbojpeg.yuv_to_jpeg(payload, width, height, state.quality, pix_fmt) do
+      {:ok, jpeg} ->
+        {[buffer: {:output, %Buffer{buffer | payload: jpeg}}], state}
+
+      error ->
+        raise """
+        could not create JPEG image
+        #{inspect(error)}
+        """
     end
   end
 end

lib/turbojpeg/jpeg_header.ex

@@ -0,0 +1,11 @@
+defmodule Turbojpeg.JpegHeader do
+  @moduledoc false
+
+  @type t :: %__MODULE__{
+          width: Turbojpeg.dimension(),
+          height: Turbojpeg.dimension(),
+          format: Turbojpeg.format()
+        }
+
+  defstruct [:width, :height, :format]
+end