Repository: elixir-grpc/grpc
Pull request #99 (2019-04-18 02:18:05)

Introduced Codec
Codecs handle encoding and decoding of gRPC messages.

Comes with two types of codecs, `GRPC.Codec.Proto`(default) and `GRPC.Codec.Erlpack`.

Currently, codecs are specified as an option on `Server` and `Stub`. 

Looking at the gRPC protocol and implementations in other languages, there should be a way to also specify a codec per request. That will require us registering multiple codecs, and being able to override a codec on the individual call, for example

```elixir
Helloworld.Greeter.Stub.reply_hello(name, codec: GRPC.Codec.Erlpack)
```

I'm still iterating on it but it would be great to get some early feedback. We've found this refactoring useful because we could start using the library without rewriting our code to use protos and we can stick with erlang serialization for a little while.



mairbek: https://github.com/elixir-grpc/grpc/issues/37 should be possible to add json and flatbuffers eventually
tony612: I saw you added codec to stream. Could we have a connection level codec also?

In addition, though I saw grpc-go supports stream level codec, but I'm thinking if we really need a stream level codec. If people don't use this often, I suggest we can only add a connection level.
mairbek: go-grpc has a `https://godoc.org/google.golang.org/grpc#WithDefaultCallOptions` and allows to override a codec during the call.

I'm going to model the code accordingly.
mairbek: I've tweaked the code a little bit

* Rather than having one codec, a server can support a list of codecs
* The server picks reads the request content type and picks the codec accordingly 
* The client can specify a `codec` option when sending a request
* The client can specify a default channel coding
tony612: @mairbek Is this finished?
mairbek: @tony612 should be good now, please take a look

Patch:
diff --git a/lib/grpc/adapter/cowboy/handler.ex b/lib/grpc/adapter/cowboy/handler.ex
index 524fff0..75b9eb1 100644
--- a/lib/grpc/adapter/cowboy/handler.ex
+++ b/lib/grpc/adapter/cowboy/handler.ex
@@ -17,35 +17,58 @@ defmodule GRPC.Adapter.Cowboy.Handler do
     path = :cowboy_req.path(req)
     server = Map.get(servers, GRPC.Server.service_name(path))
 
-    if server do
-      stream = %GRPC.Server.Stream{
-        server: server,
-        endpoint: endpoint,
-        adapter: @adapter,
-        payload: %{pid: self()},
-        local: opts[:local]
-      }
-
-      pid = spawn_link(__MODULE__, :call_rpc, [server, path, stream])
-      Process.flag(:trap_exit, true)
-
-      req = :cowboy_req.set_resp_headers(HTTP2.server_headers(), req)
-
-      timeout = :cowboy_req.header("grpc-timeout", req)
-
-      timer_ref =
-        if timeout && timeout != :undefined do
-          Process.send_after(
-            self(),
-            {:handling_timeout, self()},
-            GRPC.Transport.Utils.decode_timeout(timeout)
-          )
-        end
+    req_content_type = :cowboy_req.header("content-type", req)
+
+    codec =
+      with {:ok, subtype} <- extract_subtype(req_content_type) do
+        Enum.find(server.get_codecs(), nil, fn c -> c.content_subtype() == subtype end)
+      else
+        _ -> nil
+      end
 
-      {:cowboy_loop, req, %{pid: pid, handling_timer: timer_ref}}
+    if codec do
+      if server do
+        stream = %GRPC.Server.Stream{
+          server: server,
+          endpoint: endpoint,
+          adapter: @adapter,
+          payload: %{pid: self()},
+          local: opts[:local],
+          codec: codec
+        }
+
+        pid = spawn_link(__MODULE__, :call_rpc, [server, path, stream])
+        Process.flag(:trap_exit, true)
+
+        req = :cowboy_req.set_resp_headers(HTTP2.server_headers(stream), req)
+
+        timeout = :cowboy_req.header("grpc-timeout", req)
+
+        timer_ref =
+          if timeout && timeout != :undefined do
+            Process.send_after(
+              self(),
+              {:handling_timeout, self()},
+              GRPC.Transport.Utils.decode_timeout(timeout)
+            )
+          end
+
+        {:cowboy_loop, req, %{pid: pid, handling_timer: timer_ref}}
+      else
+        error = RPCError.new(:unimplemented)
+        trailers = HTTP2.server_trailers(error.status, error.message)
+        req = send_error_trailers(req, trailers)
+        {:ok, req, state}
+      end
     else
-      error = RPCError.new(:unimplemented)
-      trailers = HTTP2.server_trailers(error.status, error.message)
+      error = RPCError.new(:internal)
+
+      trailers =
+        HTTP2.server_trailers(
+          error.status,
+          "No codec registered for content type #{req_content_type}"
+        )
+
       req = send_error_trailers(req, trailers)
       {:ok, req, state}
     end
@@ -195,6 +218,10 @@ defmodule GRPC.Adapter.Cowboy.Handler do
     :ok
   end
 
+  def terminate(_reason, _req, _state) do
+    :ok
+  end
+
   def call_rpc(server, path, stream) do
     result =
       try do
@@ -295,4 +322,12 @@ defmodule GRPC.Adapter.Cowboy.Handler do
         end
     end
   end
+
+  defp extract_subtype("application/grpc"), do: {:ok, "proto"}
+  defp extract_subtype("application/grpc+"), do: {:ok, "proto"}
+  defp extract_subtype("application/grpc;"), do: {:ok, "proto"}
+
+  defp extract_subtype(<<"application/grpc+", rest::binary>>), do: {:ok, rest}
+  defp extract_subtype(<<"application/grpc;", rest::binary>>), do: {:ok, rest}
+  defp extract_subtype(_), do: :error
 end
diff --git a/lib/grpc/channel.ex b/lib/grpc/channel.ex
index 5df4133..8c9e468 100644
--- a/lib/grpc/channel.ex
+++ b/lib/grpc/channel.ex
@@ -12,6 +12,7 @@ defmodule GRPC.Channel do
     * `:scheme` - scheme of connection, like `http`
     * `:cred` - credentials used for authentication
     * `:adapter` - a client adapter module, like `GRPC.Adapter.Gun`
+    * `:codec` - a default codec for this channel
     * `:adapter_payload` - payload the adapter uses
   """
 
@@ -22,6 +23,7 @@ defmodule GRPC.Channel do
           cred: GRPC.Credential.t(),
           adapter: atom,
           adapter_payload: any,
+          codec: module,
           interceptors: []
         }
   defstruct host: nil,
@@ -30,5 +32,6 @@ defmodule GRPC.Channel do
             cred: nil,
             adapter: nil,
             adapter_payload: nil,
+            codec: nil,
             interceptors: []
 end
diff --git a/lib/grpc/client/stream.ex b/lib/grpc/client/stream.ex
index a07130d..438892a 100644
--- a/lib/grpc/client/stream.ex
+++ b/lib/grpc/client/stream.ex
@@ -4,17 +4,16 @@ defmodule GRPC.Client.Stream do
 
   ## Fields
 
-    * `:channel` - `GRPC.Channel`, the channel established by client
-    * `:payload` - data used by adapter in a request
-    * `:path` - the request path to sent
-    * `:marshal` - a function encoding the request
-    * `:unmarshal` - a function decoding the reply
-    * `:req_stream` - indicates if request is streaming
-    * `:res_stream` - indicates if reply is streaming
+    * `:channel`           - `GRPC.Channel`, the channel established by client
+    * `:payload`           - data used by adapter in a request
+    * `:path`              - the request path to sent
+    * `request_mod`        - the request module, or nil for untyped protocols
+    * `response_mod`       - the response module, or nil for untyped protocols
+    * `:codec`             - the codec
+    * `:req_stream`        - indicates if request is streaming
+    * `:res_stream`        - indicates if reply is streaming
   """
 
-  @typep marshal :: (struct -> binary)
-  @typep unmarshal :: (binary -> struct)
   @typep stream_payload :: any
   @type t :: %__MODULE__{
           channel: GRPC.Channel.t(),
@@ -24,12 +23,14 @@ defmodule GRPC.Client.Stream do
           rpc: tuple,
           payload: stream_payload,
           path: String.t(),
-          marshal: marshal | nil,
-          unmarshal: unmarshal | nil,
+          request_mod: atom,
+          response_mod: atom,
+          codec: map,
           server_stream: boolean,
           canceled: boolean,
           __interface__: map
         }
+
   defstruct channel: nil,
             service_name: nil,
             method_name: nil,
@@ -37,8 +38,9 @@ defmodule GRPC.Client.Stream do
             rpc: nil,
             payload: %{},
             path: nil,
-            marshal: nil,
-            unmarshal: nil,
+            request_mod: nil,
+            response_mod: nil,
+            codec: nil,
             server_stream: nil,
             # TODO: it's better to get canceled status from adapter
             canceled: false,
@@ -52,8 +54,8 @@ defmodule GRPC.Client.Stream do
 
   @doc false
   @spec send_request(GRPC.Client.Stream.t(), struct, Keyword.t()) :: GRPC.Client.Stream.t()
-  def send_request(%{marshal: marshal, channel: channel} = stream, request, opts) do
-    message = marshal.(request)
+  def send_request(%{codec: codec, channel: channel} = stream, request, opts) do
+    message = codec.encode(request)
     send_end_stream = Keyword.get(opts, :end_stream, false)
     channel.adapter.send_data(stream, message, send_end_stream: send_end_stream)
   end
diff --git a/lib/grpc/codec.ex b/lib/grpc/codec.ex
new file mode 100644
index 0000000..5789252
--- /dev/null
+++ b/lib/grpc/codec.ex
@@ -0,0 +1,10 @@
+defmodule GRPC.Codec do
+  @moduledoc """
+  Contains code to serialize a deserialize the message.
+  """
+
+  # Content subtype is a suffix after "application/grpc+" such as "proto".
+  @callback content_subtype() :: String.t()
+  @callback encode(any) :: binary
+  @callback decode(any, atom) :: any
+end
diff --git a/lib/grpc/codec/erlpack.ex b/lib/grpc/codec/erlpack.ex
new file mode 100644
index 0000000..1f00fb1
--- /dev/null
+++ b/lib/grpc/codec/erlpack.ex
@@ -0,0 +1,15 @@
+defmodule GRPC.Codec.Erlpack do
+  @behaviour GRPC.Codec
+
+  def content_subtype() do
+    "erlpack"
+  end
+
+  def encode(struct) do
+    :erlang.term_to_binary(struct)
+  end
+
+  def decode(binary, _module) do
+    :erlang.binary_to_term(binary)
+  end
+end
diff --git a/lib/grpc/codec/proto.ex b/lib/grpc/codec/proto.ex
new file mode 100644
index 0000000..afe2168
--- /dev/null
+++ b/lib/grpc/codec/proto.ex
@@ -0,0 +1,15 @@
+defmodule GRPC.Codec.Proto do
+  @behaviour GRPC.Codec
+
+  def content_subtype() do
+    "proto"
+  end
+
+  def encode(struct) do
+    Protobuf.Encoder.encode(struct)
+  end
+
+  def decode(binary, module) do
+    Protobuf.Decoder.decode(binary, module)
+  end
+end
diff --git a/lib/grpc/server.ex b/lib/grpc/server.ex
index c7e2621..0362e46 100644
--- a/lib/grpc/server.ex
+++ b/lib/grpc/server.ex
@@ -40,8 +40,14 @@ defmodule GRPC.Server do
   @type rpc :: (GRPC.Server.rpc_req(), Stream.t() -> rpc_return)
 
   defmacro __using__(opts) do
-    quote bind_quoted: [service_mod: opts[:service]] do
+    quote bind_quoted: [opts: opts] do
+      service_mod = opts[:service]
       service_name = service_mod.__meta__(:name)
+      codec = opts[:codecs] || [GRPC.Codec.Proto]
+
+      def get_codecs() do
+        unquote(codec)
+      end
 
       Enum.each(service_mod.__rpc_calls__, fn {name, _, _} = rpc ->
         func_name = name |> to_string |> Macro.underscore() |> String.to_atom()
@@ -77,14 +83,12 @@ defmodule GRPC.Server do
   @doc false
   @spec call(atom, Stream.t(), tuple, atom) :: {:ok, Stream.t(), struct} | {:ok, struct}
   def call(
-        service_mod,
+        _service_mod,
         stream,
         {_, {req_mod, req_stream}, {res_mod, res_stream}} = rpc,
         func_name
       ) do
-    marshal_func = fn res -> service_mod.marshal(res_mod, res) end
-    unmarshal_func = fn req -> service_mod.unmarshal(req_mod, req) end
-    stream = %{stream | marshal: marshal_func, unmarshal: unmarshal_func, rpc: rpc}
+    stream = %{stream | request_mod: req_mod, response_mod: res_mod, rpc: rpc}
 
     handle_request(req_stream, res_stream, stream, func_name)
   end
@@ -100,12 +104,13 @@ defmodule GRPC.Server do
   defp do_handle_request(
          false,
          res_stream,
-         %{unmarshal: unmarshal, adapter: adapter, payload: payload} = stream,
+         %{request_mod: req_mod, codec: codec, adapter: adapter, payload: payload} = stream,
          func_name
        ) do
-    {:ok, data} = adapter.read_body(payload)
+    fuf = adapter.read_body(payload)
+    {:ok, data} = fuf
     message = GRPC.Message.from_data(data)
-    request = unmarshal.(message)
+    request = codec.decode(message, req_mod)
 
     call_with_interceptors(res_stream, func_name, stream, request)
   end
@@ -113,12 +118,12 @@ defmodule GRPC.Server do
   defp do_handle_request(
          true,
          res_stream,
-         %{unmarshal: unmarshal, adapter: adapter, payload: payload} = stream,
+         %{request_mod: req_mod, codec: codec, adapter: adapter, payload: payload} = stream,
          func_name
        ) do
     reading_stream =
       adapter.reading_stream(payload)
-      |> Elixir.Stream.map(&unmarshal.(&1))
+      |> Elixir.Stream.map(fn message -> codec.decode(message, req_mod) end)
 
     call_with_interceptors(res_stream, func_name, stream, reading_stream)
   end
diff --git a/lib/grpc/server/stream.ex b/lib/grpc/server/stream.ex
index b02a401..11fd8ec 100644
--- a/lib/grpc/server/stream.ex
+++ b/lib/grpc/server/stream.ex
@@ -8,29 +8,15 @@ defmodule GRPC.Server.Stream do
 
   ## Fields
 
-    * `:server`    - user defined gRPC server module
-    * `:marshal`   - a function encoding the reply
-    * `:unmarshal` - a function decoding the request
-    * `:adapter`   - a server adapter module, like `GRPC.Adapter.Cowboy`
-    * `:payload`   - the payload needed by the adapter
-    * `:local`     - local data initialized by user
+    * `:server`            - user defined gRPC server module
+    * `:adapter`           - a server adapter module, like `GRPC.Adapter.Cowboy`
+    * `request_mod`        - the request module, or nil for untyped protocols
+    * `response_mod`       - the response module, or nil for untyped protocols
+    * `:codec`             - the codec
+    * `:payload`           - the payload needed by the adapter
+    * `:local`             - local data initialized by user
   """
 
-  defstruct server: nil,
-            service_name: nil,
-            method_name: nil,
-            grpc_type: nil,
-            endpoint: nil,
-            rpc: nil,
-            marshal: nil,
-            unmarshal: nil,
-            payload: nil,
-            adapter: nil,
-            local: nil,
-            __interface__: %{send_reply: &__MODULE__.send_reply/2}
-
-  @typep marshal :: (struct -> binary)
-  @typep unmarshal :: (binary -> struct)
   @type t :: %__MODULE__{
           server: atom,
           service_name: String.t(),
@@ -38,16 +24,31 @@ defmodule GRPC.Server.Stream do
           grpc_type: atom,
           endpoint: atom,
           rpc: tuple,
-          marshal: marshal,
-          unmarshal: unmarshal,
+          request_mod: atom,
+          response_mod: atom,
+          codec: map,
           payload: any,
           adapter: atom,
           local: any,
           __interface__: map
         }
 
-  def send_reply(%{adapter: adapter, marshal: marshal} = stream, reply) do
-    {:ok, data, _size} = reply |> marshal.() |> GRPC.Message.to_data(%{iolist: true})
+  defstruct server: nil,
+            service_name: nil,
+            method_name: nil,
+            grpc_type: nil,
+            endpoint: nil,
+            rpc: nil,
+            request_mod: nil,
+            response_mod: nil,
+            codec: nil,
+            payload: nil,
+            adapter: nil,
+            local: nil,
+            __interface__: %{send_reply: &__MODULE__.send_reply/2}
+
+  def send_reply(%{adapter: adapter, codec: codec} = stream, reply) do
+    {:ok, data, _size} = reply |> codec.encode() |> GRPC.Message.to_data(%{iolist: true})
     adapter.send_reply(stream.payload, data)
     stream
   end
diff --git a/lib/grpc/service.ex b/lib/grpc/service.ex
index d1da9ec..5357b6c 100644
--- a/lib/grpc/service.ex
+++ b/lib/grpc/service.ex
@@ -20,16 +20,6 @@ defmodule GRPC.Service do
       Module.register_attribute(__MODULE__, :rpc_calls, accumulate: true)
       @before_compile GRPC.Service
 
-      def marshal(mod, message) do
-        func = unquote(opts[:marshal]) || (&GRPC.Message.Protobuf.encode/2)
-        func.(mod, message)
-      end
-
-      def unmarshal(mod, message) do
-        func = unquote(opts[:unmarshal]) || (&GRPC.Message.Protobuf.decode/2)
-        func.(mod, message)
-      end
-
       def __meta__(:name), do: unquote(opts[:name])
     end
   end
diff --git a/lib/grpc/stub.ex b/lib/grpc/stub.ex
index 19436cd..81831c9 100644
--- a/lib/grpc/stub.ex
+++ b/lib/grpc/stub.ex
@@ -53,7 +53,8 @@ defmodule GRPC.Stub do
           | {:error, GRPC.RPCError.t()}
 
   defmacro __using__(opts) do
-    quote bind_quoted: [service_mod: opts[:service]] do
+    quote bind_quoted: [opts: opts] do
+      service_mod = opts[:service]
       service_name = service_mod.__meta__(:name)
 
       Enum.each(service_mod.__rpc_calls__, fn {name, {_, req_stream}, {_, res_stream}} = rpc ->
@@ -142,6 +143,7 @@ defmodule GRPC.Stub do
     cred = Keyword.get(opts, :cred)
     scheme = if cred, do: @secure_scheme, else: @insecure_scheme
     interceptors = Keyword.get(opts, :interceptors, []) |> init_interceptors
+    codec = Keyword.get(opts, :codec, GRPC.Codec.Proto)
 
     %Channel{
       host: host,
@@ -149,7 +151,8 @@ defmodule GRPC.Stub do
       scheme: scheme,
       cred: cred,
       adapter: adapter,
-      interceptors: interceptors
+      interceptors: interceptors,
+      codec: codec
     }
     |> adapter.connect(opts[:adapter_opts])
   end
@@ -206,20 +209,20 @@ defmodule GRPC.Stub do
       `%{headers: headers}`(server streaming)
   """
   @spec call(atom, tuple, GRPC.Client.Stream.t(), struct | nil, keyword) :: rpc_return
-  def call(service_mod, rpc, stream, request, opts) do
+  def call(_service_mod, rpc, %{channel: channel} = stream, request, opts) do
     {_, {req_mod, req_stream}, {res_mod, _}} = rpc
-    marshal = fn req -> service_mod.marshal(req_mod, req) end
-    unmarshal = fn res -> service_mod.unmarshal(res_mod, res) end
 
-    stream = %{stream | marshal: marshal, unmarshal: unmarshal}
+    stream = %{stream | request_mod: req_mod, response_mod: res_mod}
 
     opts = parse_req_opts(opts)
 
+    stream = %{stream | codec: opts[:codec] || channel.codec}
+
     do_call(req_stream, stream, request, opts)
   end
 
-  defp do_call(false, %{marshal: marshal, channel: channel} = stream, request, opts) do
-    message = marshal.(request)
+  defp do_call(false, %{codec: codec, channel: channel} = stream, request, opts) do
+    message = codec.encode(request)
 
     last = fn s, r ->
       s
@@ -368,12 +371,12 @@ defmodule GRPC.Stub do
     end
   end
 
-  def do_recv(%{payload: payload, unmarshal: unmarshal, channel: channel}, opts) do
+  def do_recv(%{payload: payload, response_mod: res_mod, codec: codec, channel: channel}, opts) do
     with {:ok, headers, _is_fin} <-
            recv_headers(channel.adapter, channel.adapter_payload, payload, opts),
          {:ok, body, trailers} <-
            recv_body(channel.adapter, channel.adapter_payload, payload, opts) do
-      {status, msg} = parse_response(body, trailers, unmarshal)
+      {status, msg} = parse_response(body, trailers, codec, res_mod)
 
       if opts[:return_headers] do
         {status, msg, %{headers: headers, trailers: trailers}}
@@ -410,13 +413,13 @@ defmodule GRPC.Stub do
     end
   end
 
-  defp parse_response(data, trailers, unmarshal) do
+  defp parse_response(data, trailers, codec, res_mod) do
     case parse_trailers(trailers) do
       :ok ->
         result =
           data
           |> GRPC.Message.from_data()
-          |> unmarshal.()
+          |> codec.decode(res_mod)
 
         {:ok, result}
 
@@ -438,7 +441,8 @@ defmodule GRPC.Stub do
   defp response_stream(
          %{
            channel: %{adapter: adapter, adapter_payload: ap},
-           unmarshal: unmarshal,
+           response_mod: res_mod,
+           codec: codec,
            payload: payload
          },
          opts
@@ -451,7 +455,8 @@ defmodule GRPC.Stub do
       fin: false,
       need_more: true,
       opts: opts,
-      unmarshal: unmarshal
+      response_mod: res_mod,
+      codec: codec
     }
 
     Stream.unfold(state, fn s -> read_stream(s) end)
@@ -500,10 +505,10 @@ defmodule GRPC.Stub do
     end
   end
 
-  defp read_stream(%{buffer: buffer, need_more: false, unmarshal: unmarshal} = s) do
+  defp read_stream(%{buffer: buffer, need_more: false, response_mod: res_mod, codec: codec} = s) do
     case GRPC.Message.get_message(buffer) do
       {message, rest} ->
-        reply = unmarshal.(message)
+        reply = codec.decode(message, res_mod)
         new_s = Map.put(s, :buffer, rest)
         {{:ok, reply}, new_s}
 
@@ -536,8 +541,8 @@ defmodule GRPC.Stub do
     parse_req_opts(t, Map.put(acc, :metadata, metadata))
   end
 
-  defp parse_req_opts([{:content_type, content_type} | t], acc) do
-    parse_req_opts(t, Map.put(acc, :content_type, content_type))
+  defp parse_req_opts([{:codec, codec} | t], acc) do
+    parse_req_opts(t, Map.put(acc, :codec, codec))
   end
 
   defp parse_req_opts([{:return_headers, return_headers} | t], acc) do
diff --git a/lib/grpc/transport/http2.ex b/lib/grpc/transport/http2.ex
index 23b26a8..74539cd 100644
--- a/lib/grpc/transport/http2.ex
+++ b/lib/grpc/transport/http2.ex
@@ -6,8 +6,8 @@ defmodule GRPC.Transport.HTTP2 do
   alias GRPC.Transport.Utils
   alias GRPC.Status
 
-  def server_headers() do
-    %{"content-type" => "application/grpc+proto"}
+  def server_headers(%{codec: codec} = _stream) do
+    %{"content-type" => "application/grpc+#{codec.content_subtype}"}
   end
 
   @spec server_trailers(integer, String.t()) :: map
@@ -29,9 +29,9 @@ defmodule GRPC.Transport.HTTP2 do
   end
 
   @spec client_headers_without_reserved(GRPC.Client.Stream.t(), map) :: [{String.t(), String.t()}]
-  def client_headers_without_reserved(_, opts \\ %{}) do
+  def client_headers_without_reserved(%{codec: codec} = _stream, opts \\ %{}) do
     [
-      {"content-type", opts[:content_type] || "application/grpc+proto"},
+      {"content-type", "application/grpc+#{codec.content_subtype}"},
       {"user-agent", "grpc-elixir/#{opts[:grpc_version] || GRPC.version()}"},
       {"te", "trailers"}
     ]


Test patch:
diff --git a/test/grpc/integration/codec_test.exs b/test/grpc/integration/codec_test.exs
new file mode 100644
index 0000000..aaa2783
--- /dev/null
+++ b/test/grpc/integration/codec_test.exs
@@ -0,0 +1,56 @@
+defmodule GRPC.Integration.CoderTest do
+  use GRPC.Integration.TestCase, async: true
+
+  defmodule NotRegisteredCodec do
+    @behaviour GRPC.Codec
+
+    def content_subtype() do
+      "not-registered"
+    end
+
+    def encode(struct) do
+      :erlang.term_to_binary(struct)
+    end
+
+    def decode(_binary, _module) do
+      :fail
+    end
+  end
+
+  defmodule HelloServer do
+    use GRPC.Server,
+      service: Helloworld.Greeter.Service,
+      codecs: [GRPC.Codec.Proto, GRPC.Codec.Erlpack]
+
+    def say_hello(req, _stream) do
+      Helloworld.HelloReply.new(message: "Hello, #{req.name}")
+    end
+  end
+
+  defmodule HelloErlpackStub do
+    use GRPC.Stub, service: Helloworld.Greeter.Service
+  end
+
+  test "Says hello over erlpack" do
+    run_server(HelloServer, fn port ->
+      {:ok, channel} = GRPC.Stub.connect("localhost:#{port}")
+      name = "Mairbek"
+      req = Helloworld.HelloRequest.new(name: name)
+
+      {:ok, reply} = channel |> HelloErlpackStub.say_hello(req, codec: GRPC.Codec.Erlpack)
+      assert reply.message == "Hello, #{name}"
+
+      # verify that proto still works
+      {:ok, reply} = channel |> HelloErlpackStub.say_hello(req, codec: GRPC.Codec.Proto)
+      assert reply.message == "Hello, #{name}"
+
+      # codec not registered
+      {:error, reply} = channel |> HelloErlpackStub.say_hello(req, codec: NotRegisteredCodec)
+
+      assert %GRPC.RPCError{
+               status: GRPC.Status.internal(),
+               message: "No codec registered for content type application/grpc+not-registered"
+             } == reply
+    end)
+  end
+end
diff --git a/test/grpc/integration/erlpack_notypes.ex b/test/grpc/integration/erlpack_notypes.ex
new file mode 100644
index 0000000..922cc04
--- /dev/null
+++ b/test/grpc/integration/erlpack_notypes.ex
@@ -0,0 +1,45 @@
+defmodule GRPC.Integration.ErplackNotypesTest do
+  use GRPC.Integration.TestCase, async: true
+
+  defmodule Helloworld.Notypes.Service do
+    use GRPC.Service, name: "helloworld.Notypes"
+
+    rpc :ReplyHello, :ignore, :ignore
+  end
+
+  defmodule HelloServer do
+    use GRPC.Server, service: Helloworld.Notypes.Service, codecs: [GRPC.Codec.Erlpack]
+
+    def reply_hello(req, _stream) do
+      {:ok, "Hello, #{req}"}
+    end
+  end
+
+  defmodule HelloErlpackStub do
+    use GRPC.Stub, service: Helloworld.Notypes.Service
+  end
+
+  test "Says hello over erlpack" do
+    run_server(HelloServer, fn port ->
+      {:ok, channel} =
+        GRPC.Stub.connect(
+          "localhost:#{port}",
+          interceptors: [GRPC.Logger.Client],
+          codec: GRPC.Codec.Erlpack
+        )
+
+      name = "World"
+      {:ok, reply} = channel |> HelloErlpackStub.reply_hello(name)
+      assert reply == {:ok, "Hello, #{name}"}
+    end)
+  end
+
+  test "Says hello over erlpack call level" do
+    run_server(HelloServer, fn port ->
+      {:ok, channel} = GRPC.Stub.connect("localhost:#{port}", interceptors: [GRPC.Logger.Client])
+      name = "World"
+      {:ok, reply} = channel |> HelloErlpackStub.reply_hello(name, codec: GRPC.Codec.Erlpack)
+      assert reply == {:ok, "Hello, #{name}"}
+    end)
+  end
+end
diff --git a/test/grpc/server_test.exs b/test/grpc/server_test.exs
index c06a8da..ecea03f 100644
--- a/test/grpc/server_test.exs
+++ b/test/grpc/server_test.exs
@@ -15,7 +15,7 @@ defmodule GRPC.ServerTest do
   end
 
   test "send_reply/2 works" do
-    stream = %GRPC.Server.Stream{adapter: GRPC.Test.ServerAdapter, marshal: & &1}
+    stream = %GRPC.Server.Stream{adapter: GRPC.Test.ServerAdapter, codec: GRPC.Codec.Erlpack}
     response = <<1, 2, 3, 4, 5, 6, 7, 8>>
     assert %GRPC.Server.Stream{} = GRPC.Server.send_reply(stream, response)
   end
diff --git a/test/grpc/transport/http2_test.exs b/test/grpc/transport/http2_test.exs
index 235eb84..4b5d935 100644
--- a/test/grpc/transport/http2_test.exs
+++ b/test/grpc/transport/http2_test.exs
@@ -4,9 +4,10 @@ defmodule GRPC.Transport.HTTP2Test do
   alias GRPC.Transport.HTTP2
 
   @channel %Channel{scheme: "http", host: "grpc.io"}
+  @codec GRPC.Codec.Proto
 
   test "client_headers/3 returns basic headers" do
-    stream = %{channel: @channel, path: "/foo/bar"}
+    stream = %{channel: @channel, path: "/foo/bar", codec: @codec}
     headers = HTTP2.client_headers(stream, %{grpc_version: "1.0.0"})
 
     assert headers == [
@@ -21,19 +22,19 @@ defmodule GRPC.Transport.HTTP2Test do
   end
 
   test "client_headers/3 returns grpc-encoding" do
-    stream = %{channel: @channel, path: "/foo/bar"}
+    stream = %{channel: @channel, path: "/foo/bar", codec: @codec}
     headers = HTTP2.client_headers(stream, %{grpc_encoding: "gzip"})
     assert List.last(headers) == {"grpc-encoding", "gzip"}
   end
 
   test "client_headers/3 returns custom metadata" do
-    stream = %{channel: @channel, path: "/foo/bar"}
+    stream = %{channel: @channel, path: "/foo/bar", codec: @codec}
     headers = HTTP2.client_headers(stream, %{metadata: %{foo: "bar", foo1: :bar1}})
     assert [{"foo1", "bar1"}, {"foo", "bar"} | _] = Enum.reverse(headers)
   end
 
   test "client_headers/3 returns custom metadata with *-bin key" do
-    stream = %{channel: @channel, path: "/foo/bar"}
+    stream = %{channel: @channel, path: "/foo/bar", codec: @codec}
 
     headers =
       HTTP2.client_headers(stream, %{metadata: %{"key1-bin" => "abc", "key2-bin" => <<194, 128>>}})
@@ -42,7 +43,7 @@ defmodule GRPC.Transport.HTTP2Test do
   end
 
   test "client_headers/3 rejects reserved headers in metadata" do
-    stream = %{channel: @channel, path: "/foo/bar"}
+    stream = %{channel: @channel, path: "/foo/bar", codec: @codec}
 
     metadata = %{
       "foo" => "bar",
@@ -57,27 +58,30 @@ defmodule GRPC.Transport.HTTP2Test do
   end
 
   test "client_headers/3 downcase keys of metadata" do
-    stream = %{channel: @channel, path: "/foo/bar"}
+    stream = %{channel: @channel, path: "/foo/bar", codec: @codec}
     metadata = %{:Foo => "bar", "Foo-Bar" => "bar"}
     headers = HTTP2.client_headers(stream, %{metadata: metadata})
     assert [{"foo-bar", "bar"}, {"foo", "bar"} | _] = Enum.reverse(headers)
   end
 
   test "client_headers/3 merges metadata with same keys" do
-    stream = %{channel: @channel, path: "/foo/bar"}
+    stream = %{channel: @channel, path: "/foo/bar", codec: @codec}
     headers = HTTP2.client_headers(stream, %{metadata: [foo: "bar", foo: :bar1]})
     assert [{"foo", "bar,bar1"} | _] = Enum.reverse(headers)
   end
 
   test "client_headers/3 has timeout with :timeout option" do
-    stream = %{channel: @channel, path: "/foo/bar"}
+    stream = %{channel: @channel, path: "/foo/bar", codec: @codec}
     headers = HTTP2.client_headers(stream, %{timeout: 5})
     assert [{"grpc-timeout", "5m"} | _] = Enum.reverse(headers)
   end
 
   test "client_headers/3 support custom content-type" do
-    stream = %{channel: @channel, path: "/foo/bar"}
-    headers = HTTP2.client_headers(stream, %{content_type: "application/grpc"})
-    assert {_, "application/grpc"} = Enum.find(headers, fn {key, _} -> key == "content-type" end)
+    # TODO mairbek figure otu
+    stream = %{channel: @channel, path: "/foo/bar", codec: %{content_subtype: "proto"}}
+    headers = HTTP2.client_headers(stream, %{})
+
+    assert {_, "application/grpc+proto"} =
+             Enum.find(headers, fn {key, _} -> key == "content-type" end)
   end
 end