Protocol v4 and v3

.travis.yml

@@ -0,0 +1,23 @@
+language: erlang
+sudo: required
+# Install cassandra
+# Standard Travis cassandra is 2.0.9
+# Docker runs on Trusty planform (the new one)
+install:
+    - docker run -d -p 9042:9042 -e MAX_HEAP_SIZE=128M -e HEAP_NEWSIZE=64M --name=cassandra cassandra:${CASSANDRA_VERSION}
+    - ./wait_for_cassandra.sh || docker logs cassandra
+script:
+    - ./rebar skip_deps=true eunit -v
+
+services:
+        - docker
+
+otp_release:
+    - R16B03
+    - 17.5
+    - 18.1
+
+env:
+    - CASSANDRA_VERSION=2.1
+    - CASSANDRA_VERSION=3.1
+    - CASSANDRA_VERSION=3.2

include/builtin_types.hrl

@@ -5,5 +5,5 @@
 -else.
     -type dict_t() :: dict:dict().
     -type queue_t() :: queue:queue().
-    -type set_t() :: set:set().
+    -type set_t() :: sets:set().
 -endif.

include/constants.hrl

@@ -27,6 +27,9 @@
 -define(TRUNCATE_ERROR,    16#1003).
 -define(WRITE_TIMEOUT,     16#1100).
 -define(READ_TIMEOUT,      16#1200).
+-define(READ_FAILURE,      16#1300).
+-define(FUNCTION_FAILURE,  16#1400).
+-define(WRITE_FAILURE,     16#1500).
 
 %% 2xx: problem validating the request
 -define(SYNTAX_ERROR,      16#2000).

src/seestar_cqltypes.erl

@@ -101,36 +101,36 @@ decode_value(_Type, Bytes) ->
 %% -------------------------------------------------------------------------
 
 decode_list(Type, Data) ->
-    {Length, Rest} = seestar_types:decode_short(Data),
+    {Length, Rest} = seestar_types:decode_int(Data),
     decode_list(Type, Length, Rest, []).
 
 decode_list(_, 0, _, List) ->
     lists:reverse(List);
 decode_list(Type, Length, Data, List) ->
-    {Bytes, Rest} = seestar_types:decode_short_bytes(Data),
+    {Bytes, Rest} = seestar_types:decode_bytes(Data),
     decode_list(Type, Length - 1, Rest, [decode_value(Type, Bytes)|List]).
 
 decode_map(KeyType, ValueType, Data) ->
-    {Size, Rest} = seestar_types:decode_short(Data),
+    {Size, Rest} = seestar_types:decode_int(Data),
     decode_map(KeyType, ValueType, Size, Rest, dict:new()).
 
 decode_map(_, _, 0, _, Dict) ->
     Dict;
 decode_map(KeyType, ValueType, Size, Data, Dict) ->
-    {KeyBytes, Rest0} = seestar_types:decode_short_bytes(Data),
-    {ValueBytes, Rest1} = seestar_types:decode_short_bytes(Rest0),
+    {KeyBytes, Rest0} = seestar_types:decode_bytes(Data),
+    {ValueBytes, Rest1} = seestar_types:decode_bytes(Rest0),
     Key = decode_value(KeyType, KeyBytes),
     Value = decode_value(ValueType, ValueBytes),
     decode_map(KeyType, ValueType, Size - 1, Rest1, dict:store(Key, Value, Dict)).
 
 decode_set(Type, Data) ->
-    {Size, Rest} = seestar_types:decode_short(Data),
+    {Size, Rest} = seestar_types:decode_int(Data),
     decode_set(Type, Size, Rest, sets:new()).
 
 decode_set(_, 0, _, Set) ->
     Set;
 decode_set(Type, Size, Data, Set) ->
-    {Bytes, Rest} = seestar_types:decode_short_bytes(Data),
+    {Bytes, Rest} = seestar_types:decode_bytes(Data),
     decode_set(Type, Size - 1, Rest, sets:add_element(decode_value(Type, Bytes), Set)).
 
 %% -------------------------------------------------------------------------
@@ -212,23 +212,23 @@ encode_varint_neg(X, Ds) ->
 %% -------------------------------------------------------------------------
 
 encode_list(Type, List) ->
-    Encoded = [ seestar_types:encode_short_bytes(encode_value(Type, V)) || V <- List ],
-    <<(seestar_types:encode_short(length(List)))/binary,
+    Encoded = [ seestar_types:encode_bytes(encode_value(Type, V)) || V <- List ],
+    <<(seestar_types:encode_int(length(List)))/binary,
       (list_to_binary(Encoded))/binary>>.
 
 encode_set(Type, Set) ->
     Encoded = sets:fold(fun(V, Acc) ->
-                            [seestar_types:encode_short_bytes(encode_value(Type, V))|Acc]
+                            [seestar_types:encode_bytes(encode_value(Type, V))|Acc]
                         end, [], Set),
-    <<(seestar_types:encode_short(sets:size(Set)))/binary,
+    <<(seestar_types:encode_int(sets:size(Set)))/binary,
       (list_to_binary(Encoded))/binary>>.
 
 encode_map(KeyType, ValueType, Dict) ->
     F = fun(K, V, Acc) ->
-            [<<(seestar_types:encode_short_bytes(encode_value(KeyType, K)))/binary,
-               (seestar_types:encode_short_bytes(encode_value(ValueType, V)))/binary>>|Acc]
+            [<<(seestar_types:encode_bytes(encode_value(KeyType, K)))/binary,
+               (seestar_types:encode_bytes(encode_value(ValueType, V)))/binary>>|Acc]
          end,
-    <<(seestar_types:encode_short(dict:size(Dict)))/binary,
+    <<(seestar_types:encode_int(dict:size(Dict)))/binary,
       (list_to_binary(dict:fold(F, [], Dict)))/binary>>.
 
 %% -------------------------------------------------------------------------
@@ -254,9 +254,24 @@ decode_type(Data) ->
             {{map, KeyType, ValueType}, Rest2};
         16#22 ->
             {Subtype, Rest1} = decode_type(Rest0),
-            {{set, Subtype}, Rest1}
+            {{set, Subtype}, Rest1};
+        16#30 ->
+            erlang:error({todo, udt});
+        16#31 ->
+            {N, Rest1} = seestar_types:decode_short(Rest0),
+            {Types, Rest2} = decode_types(N, Rest1),
+            {{type, Types}, Rest2}
     end.
 
+decode_types(N, Data) ->
+    decode_types(N, Data, []).
+
+decode_types(0, Data, Types) ->
+    {lists:reverse(Types), Data};
+decode_types(N, Data, Types) when N > 0 ->
+    {Type, Rest0} = decode_type(Data),
+    decode_types(N-1, Rest0, [Type|Types]).
+
 code_to_type(16#01) -> ascii;
 code_to_type(16#02) -> bigint;
 code_to_type(16#03) -> blob;
@@ -272,4 +287,8 @@ code_to_type(16#0C) -> uuid;
 code_to_type(16#0D) -> varchar;
 code_to_type(16#0E) -> varint;
 code_to_type(16#0F) -> timeuuid;
-code_to_type(16#10) -> inet.
+code_to_type(16#10) -> inet;
+code_to_type(16#11) -> date;
+code_to_type(16#12) -> time;
+code_to_type(16#13) -> smallint;
+code_to_type(16#14) -> tinyint.

src/seestar_frame.erl

@@ -15,8 +15,8 @@
 %%% @private
 -module(seestar_frame).
 
--export([new/4, id/1, flags/1, has_flag/2, opcode/1,
-         body/1, encode/1, pending_size/1, decode/1]).
+-export([new/5, id/1, flags/1, has_flag/2, opcode/1,
+         body/1, warnings/1, encode/1, pending_size/1, decode/1]).
 
 -type stream_id() :: -1..127.
 -type flag() :: compression | tracing.
@@ -25,20 +25,28 @@
 
 -define(COMPRESSION, 16#01).
 -define(TRACING, 16#02).
+-define(WARNING, 16#08).
 
--record(frame, {id :: stream_id(),
+-define(REQ_VSN3,  16#03).
+-define(RESP_VSN3, 16#83).
+-define(REQ_VSN4,  16#04).
+-define(RESP_VSN4, 16#84).
+
+-record(frame, {proto,
+                id :: stream_id(),
                 flags = [] :: [flag()],
                 opcode :: opcode(),
-                body :: binary()}).
+                body :: binary(),
+                warnings = [] :: [binary()]}).
 -opaque frame() :: #frame{}.
 
 %% -------------------------------------------------------------------------
 %% API
 %% -------------------------------------------------------------------------
 
--spec new(stream_id(), [flag()], opcode(), binary()) -> frame().
-new(ID, Flags, Op, Body) ->
-    #frame{id = ID, flags = Flags, opcode = Op, body = Body}.
+-spec new(integer(), stream_id(), [flag()], opcode(), binary()) -> frame().
+new(ProtoVsn, ID, Flags, Op, Body) ->
+    #frame{proto=ProtoVsn, id = ID, flags = Flags, opcode = Op, body = Body}.
 
 -spec id(frame()) -> stream_id().
 id(Frame) ->
@@ -60,9 +68,13 @@ opcode(Frame) ->
 body(Frame) ->
     Frame#frame.body.
 
+-spec warnings(frame()) -> list(binary()).
+warnings(Frame) ->
+    Frame#frame.warnings.
+
 -spec encode(frame()) -> binary().
-encode(#frame{id = ID, flags = Flags, opcode = Op, body = Body}) ->
-    <<16#01, (encode_flags(Flags)), ID/signed, Op, (size(Body)):32, Body/binary>>.
+encode(#frame{proto=ProtoVsn, id = ID, flags = Flags, opcode = Op, body = Body}) ->
+    <<ProtoVsn, (encode_flags(Flags)), ID:16/signed, Op, (size(Body)):32, Body/binary>>.
 
 encode_flags(Flags) ->
     lists:foldl(fun(Flag, Byte) -> encode_flag(Flag) bor Byte end, 0, Flags).
@@ -71,27 +83,40 @@ encode_flag(compression) -> ?COMPRESSION;
 encode_flag(tracing)     -> ?TRACING.
 
 -spec pending_size(binary()) -> pos_integer().
-pending_size(<<16#81, _Flags, _ID/signed, _Op, Size:32, _/binary>>) ->
+pending_size(<<16#84, _Flags, _ID:16/signed, _Op, Size:32, _/binary>>) ->
     Size + 8;
 pending_size(_) ->
     undefined.
 
 -spec decode(binary()) -> {[frame()], binary()}.
 decode(Stream) ->
+    assert_protocol_version(Stream),
     decode(Stream, []).
-decode(<<16#81, Flags, ID/signed, Op, Size:32, Body:Size/binary, Rest/binary>>, Acc) ->
-    Frame = #frame{id = ID, flags = decode_flags(Flags), opcode = Op, body = Body},
+decode(<<_ProtoVsn, Flags, ID:16/signed, Op, Size:32, Body:Size/binary, Rest/binary>>, Acc) ->
+    {Warnings, Body2} = maybe_decode_warning(Flags, Body),
+    Frame = #frame{id = ID, flags = decode_flags(Flags), opcode = Op, body = Body2, warnings = Warnings},
     decode(Rest, [Frame|Acc]);
 decode(Stream, Acc) ->
     {lists:reverse(Acc), Stream}.
 
+assert_protocol_version(<<?RESP_VSN3, _/binary>>) -> ok;
+assert_protocol_version(<<?RESP_VSN4, _/binary>>) -> ok;
+assert_protocol_version(<<>>)                     -> ok.
+
+maybe_decode_warning(Flags, Body) when Flags band ?WARNING =:= ?WARNING ->
+    %% Warning flag set
+    seestar_types:decode_string_list(Body);
+maybe_decode_warning(_Flags, Body) ->
+    {[], Body}.
+
 decode_flags(Byte) ->
     F = fun(Mask, Flags) when Byte band Mask =:= Mask ->
             [decode_flag(Mask)|Flags];
            (_Mask, Flags) ->
             Flags
         end,
-    lists:foldl(F, [], [?COMPRESSION, ?TRACING]).
+    lists:foldl(F, [], [?COMPRESSION, ?TRACING, ?WARNING]).
 
 decode_flag(?COMPRESSION) -> compression;
-decode_flag(?TRACING)     -> tracing.
+decode_flag(?TRACING)     -> tracing;
+decode_flag(?WARNING)     -> warning.