snappy compression

smarin · Jan 27, 2016 · 7096f80 · 7096f80
1 parent cde146d
commit 7096f80
Show file tree

Hide file tree

Showing 2 changed files with 8 additions and 5 deletions.
diff --git a/rebar.config b/rebar.config
@@ -3,6 +3,7 @@
 {cover_print_enabled, true}.
 {clean_files, ["ebin/*.beam"]}.
 {deps,[
-       {gproc, ".*", {git, "https://github.com/uwiger/gproc.git"}}
+       {gproc, ".*", {git, "https://github.com/uwiger/gproc.git"}},
+       {esnappy, ".*", {git, "https://github.com/thekvs/esnappy.git"}}
 ]}.
 {sub_dirs, ["src"]}.
diff --git a/src/ekaf_protocol_produce.erl b/src/ekaf_protocol_produce.erl
@@ -96,13 +96,15 @@ encode_partition(Partition) when is_integer(Partition)->
 encode_partition(#partition{ id = Id,  message_sets = MessageSets })->
     {ok, Codec} = ekaf_lib:get_compression_codec(),
     MessageSetsEncoded = case Codec of
-        gzip->
+        gzip ->
             MsgVal = zlib:gzip(encode_message_sets(MessageSets)),
             SecondMsgSet = #message_set{size = 1, messages = [#message{attributes = 1, key = undefined, value = MsgVal}]},
             encode_message_set(SecondMsgSet);
-        snappy->
-            %not implementing yet because I don't need it
-            encode_message_sets(MessageSets);
+        snappy ->
+            {ok, Ctx} = esnappy:create_ctx(),
+            MsgVal = esnappy:compress(Ctx, encode_message_sets(MessageSets)),
+            SecondMsgSet = #message_set{size = 1, messages = [#message{attributes = 2, key = undefined, value = MsgVal}]},
+            encode_message_sets(SecondMsgSet);
         _ ->
             encode_message_sets(MessageSets)
     end,