Merge pull request #8 from lpgauth/dev

0.2.2

rebar.config

@@ -3,7 +3,7 @@
 
 {deps, [
   {edown, ".*", {git, "https://github.com/esl/edown.git", "HEAD"}},
-  {lz4, ".*", {git, "https://github.com/lpgauth/erlang-lz4.git", {branch, "marina"}}},
+  {lz4, ".*", {git, "https://github.com/szktty/erlang-lz4.git", "0572b0ea2a"}},
   {timing, ".*", {git, "https://github.com/lpgauth/timing.git", {branch, "master"}}}
 ]}.
 

src/marina.app.src

@@ -1,6 +1,6 @@
 {application, marina, [
   {description, "cassandra client"},
-  {vsn, "0.2.1"},
+  {vsn, "0.2.2"},
   {registered, []},
   {applications, [
     kernel,

src/marina.erl

@@ -20,43 +20,43 @@
 
 %% public
 -spec async_execute(statement_id(), consistency(), [flag()], pid()) ->
-    {ok, erlang:ref()} | {error, backlog_full}.
+    {ok, reference()} | {error, backlog_full}.
 
 async_execute(StatementId, ConsistencyLevel, Flags, Pid) ->
     async_execute(StatementId, [], ConsistencyLevel, Flags, Pid).
 
 -spec async_execute(statement_id(), [value()], consistency(), [flag()], pid()) ->
-    {ok, erlang:ref()} | {error, backlog_full}.
+    {ok, reference()} | {error, backlog_full}.
 
 async_execute(StatementId, Values, ConsistencyLevel, Flags, Pid) ->
     async_call({execute, StatementId, Values, ConsistencyLevel, Flags}, Pid).
 
 -spec async_prepare(query(), pid()) ->
-    {ok, erlang:ref()} | {error, backlog_full}.
+    {ok, reference()} | {error, backlog_full}.
 
 async_prepare(Query, Pid) ->
     async_call({prepare, Query}, Pid).
 
 -spec async_query(query(), consistency(), [flag()], pid()) ->
-    {ok, erlang:ref()} | {error, backlog_full}.
+    {ok, reference()} | {error, backlog_full}.
 
 async_query(Query, ConsistencyLevel, Flags, Pid) ->
     async_query(Query, [], ConsistencyLevel, Flags, Pid).
 
 -spec async_query(query(), [value()], consistency(), [flag()], pid()) ->
-    {ok, erlang:ref()} | {error, backlog_full}.
+    {ok, reference()} | {error, backlog_full}.
 
 async_query(Query, Values, ConsistencyLevel, Flags, Pid) ->
     async_call({query, Query, Values, ConsistencyLevel, Flags}, Pid).
 
 -spec async_reusable_query(query(), consistency(), [flag()], pid(), timeout()) ->
-    {ok, erlang:ref()} | {error, term()}.
+    {ok, reference()} | {error, term()}.
 
 async_reusable_query(Query, ConsistencyLevel, Flags, Pid, Timeout) ->
     async_reusable_query(Query, [], ConsistencyLevel, Flags, Pid, Timeout).
 
 -spec async_reusable_query(query(), [value()], consistency(), [flag()], pid(), timeout()) ->
-    {ok, erlang:ref()} | {error, term()}.
+    {ok, reference()} | {error, term()}.
 
 async_reusable_query(Query, Values, ConsistencyLevel, Flags, Pid, Timeout) ->
     case marina_cache:get(Query) of

src/marina_frame.erl

@@ -50,7 +50,7 @@ pending_size(_) ->
 decode_body(0, Body) ->
     Body;
 decode_body(1, Body) ->
-    {ok, Body2} = lz4:unpack(Body),
+    {ok, Body2} = marina_utils:unpack(Body),
     Body2.
 
 decode(<<1:1, ?PROTO_VERSION:7/unsigned-integer, Flags:8/unsigned-integer,
@@ -70,5 +70,5 @@ decode(Rest, Acc) ->
 encode_body(0, Body) ->
     Body;
 encode_body(1, Body) ->
-    {ok, Body2} = lz4:pack(iolist_to_binary(Body)),
+    {ok, Body2} = marina_utils:pack(Body),
     Body2.

src/marina_request.erl

@@ -30,7 +30,7 @@ prepare(Stream, FrameFlags, Query) ->
         stream = Stream,
         opcode = ?OP_PREPARE,
         flags = frame_flags(FrameFlags),
-        body = [<<(marina_types:encode_long_string(Query))/binary>>]
+        body = [marina_types:encode_long_string(Query)]
     }).
 
 -spec query(stream(), [frame_flag()], query(), [value()], consistency(), [flag()]) -> iolist().
@@ -41,9 +41,9 @@ query(Stream, FrameFlags, Query, Values, ConsistencyLevel, Flags) ->
         stream = Stream,
         opcode = ?OP_QUERY,
         flags = frame_flags(FrameFlags),
-        body = [<<(marina_types:encode_long_string(Query))/binary,
-            (marina_types:encode_short(ConsistencyLevel))/binary, Flags2,
-            Values2/binary>>]
+        body = [marina_types:encode_long_string(Query),
+            marina_types:encode_short(ConsistencyLevel), Flags2,
+            Values2]
     }).
 
 -spec startup([frame_flag()]) -> iolist().
@@ -77,8 +77,7 @@ flags_and_values(Flags, Values) ->
     Flags2 = flags([{values, true} | Flags]),
     ValuesCount = length(Values),
     EncodedValues = [marina_types:encode_bytes(Value) || Value <- Values],
-    Values2 = <<(marina_types:encode_short(ValuesCount))/binary,
-        (iolist_to_binary(EncodedValues))/binary>>,
+    Values2 = [marina_types:encode_short(ValuesCount), EncodedValues],
     {Flags2, Values2}.
 
 frame_flags([]) ->

src/marina_sup.erl

@@ -23,5 +23,5 @@ init([]) ->
     marina_queue:init(),
 
     {ok, {{one_for_one, 5, 10},
-        marina_utils:childs_specs()
+        marina_utils:child_specs()
     }}.

src/marina_utils.erl

@@ -2,32 +2,61 @@
 -include("marina.hrl").
 
 -export([
-    childs_specs/0,
+    child_specs/0,
     child_name/1,
     info_msg/2,
+    pack/1,
     timeout/2,
+    unpack/1,
     warning_msg/2
 
 ]).
 
 %% public
-childs_specs() ->
+-spec child_specs() -> [supervisor:child_spec()].
+
+child_specs() ->
     PoolSize = application:get_env(?APP, pool_size, ?DEFAULT_POOL_SIZE),
     [child_spec(N) || N <- lists:seq(1, PoolSize)].
 
+-spec child_name(integer()) -> atom().
+
 child_name(N) ->
     list_to_atom(?SERVER_BASE_NAME ++ integer_to_list(N)).
 
+-spec info_msg(string(), [term()]) -> ok.
+
 info_msg(Format, Data) ->
     error_logger:info_msg(Format, Data).
 
+-spec pack(binary() | iolist()) -> {ok, binary()} | {error, term()}.
+
+pack(Iolist) when is_list(Iolist) ->
+    pack(iolist_to_binary(Iolist));
+pack(Binary) ->
+    case lz4:compress(Binary, []) of
+        {ok, Compressed} ->
+            {ok, <<(size(Binary)):32/unsigned-integer, Compressed/binary>>};
+        Error ->
+            Error
+    end.
+
+-spec timeout(erlang:timestamp(), non_neg_integer()) -> non_neg_integer().
+
 timeout(Timestamp, Timeout) ->
     Diff = timer:now_diff(os:timestamp(), Timestamp) div 1000,
     case Timeout - Diff of
         X when X < 0 -> 0;
         X -> X
     end.
 
+-spec unpack(binary()) -> {ok, binary()} | {error, term()}.
+
+unpack(<<Size:32/unsigned-integer, Binary/binary>>) ->
+    lz4:uncompress(Binary, Size).
+
+-spec warning_msg(string(), [term()]) -> ok.
+
 warning_msg(Format, Data) ->
     error_logger:warning_msg(Format, Data).