Add support for batch request

Right now it only supports prepared queries.
Bump to 0.2.0-pre1.

src/erlcql.app.src

@@ -1,6 +1,6 @@
 {application, erlcql,
  [{description, "Cassandra native protocol CQL client"},
-  {vsn, "0.1.7"},
+  {vsn, "0.2.0"},
   {registered, []},
   {applications, [kernel,
                   stdlib,

src/erlcql_client.erl

@@ -111,6 +111,8 @@ prepare(Pid, QueryString, Name) ->
 execute(Pid, QueryId, Values, Consistency) ->
     async_call(Pid, {execute, QueryId, Values, Consistency}).
 
+-spec batch(pid(), [{atom(), values()}], proplist()) ->
+          result() | {error, Reason :: term()}.
 batch(Pid, Queries, Params) ->
     async_call(Pid, {batch, Queries, Params}).
 
@@ -258,7 +260,7 @@ startup({_Ref, {prepare, Query, _}}, _From, State) ->
     not_ready(prepare, Query, State);
 startup({_Ref, {execute, Query, _, _}}, _From, State) ->
     not_ready(execute, Query, State);
-startup({_Ref, {batch, _, _, _}}, _From, State) ->
+startup({_Ref, {batch, _, _}}, _From, State) ->
     not_ready(batch, State);
 startup({_Ref, options}, _From, State) ->
     not_ready(options, State);
@@ -328,9 +330,15 @@ ready({Ref, {execute, Name, Values, Consistency}}, {From, _},
             {reply, {error, invalid_query_name}, ready, State}
     end;
 ready({Ref, {batch, Queries, Params}}, {From, _},
-       #state{version = Version} = State) ->
-    Batch = erlcql_encode:batch(Version, Queries, Params),
-    send(Batch, {Ref, From}, State);
+       #state{prepared_ets = PreparedETS,
+              version = Version} = State) ->
+    case expand_prepared(PreparedETS, Queries, []) of
+        {error, Error} ->
+            {reply, {error, Error}, ready, State};
+        Queries2 ->
+            Batch = erlcql_encode:batch(Version, Queries2, Params),
+            send(Batch, {Ref, From}, State)
+    end;
 ready({Ref, options}, {From, _}, #state{version = Version} = State) ->
     Options = erlcql_encode:options(Version),
     send(Options, {Ref, From}, State);
@@ -385,6 +393,21 @@ code_change(_OldVsn, StateName, State, _Extra) ->
 terminate(_Reason, _StateName, #state{socket = Socket}) ->
     close_socket(Socket).
 
+-spec expand_prepared(ets(), [{atom(), values()}], [{binary(), values()}]) ->
+          [{binary(), values()}] | {error, invalid_query_name}.
+expand_prepared(_, [], Acc) ->
+    lists:reverse(Acc);
+expand_prepared(ETS, [{Name, Values} | Qs], Acc) when is_atom(Name) ->
+    case ets:lookup(ETS, Name) of
+        [{Name, _Query, QueryId, undefined}] ->
+            expand_prepared(ETS, Qs, [{QueryId, Values} | Acc]);
+        [{Name, _Query, QueryId, Types}] ->
+            TypedValues = lists:zip(Types, Values),
+            expand_prepared(ETS, Qs, [{QueryId, TypedValues} | Acc]);
+        [] ->
+            {error, invalid_query_name}
+    end.
+
 -spec maybe_create_prepared_ets(proplist()) -> ets().
 maybe_create_prepared_ets(Opts) ->
     New = fun() -> ets:new(?PREPARED_ETS_NAME, ?PREPARED_ETS_OPTS) end,

src/erlcql_encode.erl

@@ -132,10 +132,8 @@ batch(2, Queries, Params) when is_list(Queries) ->
     {batch, [batch_type(BatchType), short(length(Queries)), Queries2,
              short(consistency(Consistency))]}.
 
-batch_query({Query, Values}) when is_atom(Query) ->
-    [1, short_bytes(Query), erlcql_convert:to_binary(Values)];
-batch_query({Query, Values}) ->
-    [1, long_string(Query), erlcql_convert:to_binary(Values)].
+batch_query({QueryId, Values}) ->
+    [1, short_bytes(QueryId), erlcql_convert:to_binary(Values)].
 
 -spec register(version(), [event_type()]) -> {register, iolist()}.
 register(_V, Events) ->

test/erlcql_prepare_SUITE.erl

@@ -8,7 +8,8 @@
 %% Suite ----------------------------------------------------------------------
 
 all() ->
-    [prepare_execute].
+    [prepare_execute,
+     prepare_batch].
 
 %% Fixtures -------------------------------------------------------------------
 
@@ -68,3 +69,26 @@ prepare_execute(Config) ->
 
     {[Row], _} = execute(Client, select, [hd(Values)]),
     Row = Values.
+
+prepare_batch(Config) ->
+    Keyspace = ?c(keyspace, Config),
+    Table = gen_table_name(),
+    Create = [<<"CREATE TABLE ">>, Keyspace, <<".">>, Table,
+              <<" (x int PRIMARY KEY, y int)">>],
+    single_query(Create),
+
+    Insert = [<<"INSERT INTO ">>, Table,
+              <<" (x, y) VALUES (?, ?)">>],
+    Select = [<<"SELECT * FROM ">>, Table],
+    Prepare = [{insert, Insert},
+               {select, Select}],
+    Opts = [{use, Keyspace},
+            {prepare, Prepare}],
+    Client = start_client(Opts),
+
+    Points = [[1, 1], [2, 3], [3, 5]],
+    Queries = [{insert, V} || V <- Points],
+    batch(Client, Queries),
+
+    {Rows, _} = execute(Client, select, []),
+    Points = lists:sort(Rows).

test/erlcql_test.erl

@@ -62,6 +62,11 @@ execute(Pid, Name, Values) ->
     {ok, Response} = erlcql_client:execute(Pid, Name, Values, ?CONSISTENCY),
     Response.
 
+batch(Pid, Queries) ->
+    Opts = [{batch_type, logged} | ?CONSISTENCY],
+    {ok, Response} = erlcql_client:batch(Pid, Queries, Opts),
+    Response.
+
 -spec stop_client(pid()) -> ok.
 stop_client(Pid) ->
     true = unlink(Pid),

test/erlcql_test.hrl

@@ -5,6 +5,7 @@
                       single_query/1,
                       'query'/2,
                       execute/3,
+                      batch/2,
                       start_client/1,
                       stop_client/1]).