cryptix/caberl - Change PFC6VDDZ55HDVMZYGJJOBUA4PTKDYIOBV7GKHT2FCWUJGXHXAPCAC

add message chunking

Created by Henry on October 14, 2023

PFC6VDDZ55HDVMZYGJJOBUA4PTKDYIOBV7GKHT2FCWUJGXHXAPCAC

Dependencies

In channels

main

Change contents

Replacement in test/cable_basic_SUITE.erl at line 11 [5.22]

B:BD[3.88] → [4.2:72]

    {ok, [Header,  [{ttl, 1}, {hashes, Hashes}]]} = wire:decode(Bin),

[3.88]

[6.88]

    {ok, [Msg]} = wire:split_messages(Bin),
    {ok, Decoded} = wire:decode(Msg),
    [Header,  [{ttl, 1}, {hashes, Hashes}]] = Decoded,

Replacement in test/cable_basic_SUITE.erl at line 22 [5.22]

B:BD[7.542] → [4.73:147]

    {ok, [Header, [ {ttl, 1}, {cancelid, CancelId}]]} = wire:decode(Bin),

[7.542]

[6.235]

    {ok, [Msg]} = wire:split_messages(Bin),
    {ok, Decoded} = wire:decode(Msg),
    [Header, [ {ttl, 1}, {cancelid, CancelId}]] = Decoded,

Replacement in test/cable_basic_SUITE.erl at line 31 [5.22]

B:BD[7.1174] → [4.148:186]

    {ok, Decoded} = wire:decode(Bin),

[7.1174]

[4.186]

    {ok, [Msg]} = wire:split_messages(Bin),
    {ok, Decoded} = wire:decode(Msg),

Replacement in test/cable_basic_SUITE.erl at line 48 [5.22]

B:BD[6.768] → [4.231:269]

    {ok, Decoded} = wire:decode(Bin),

[6.768]

[4.269]

    {ok, [Msg]} = wire:split_messages(Bin),
    {ok, Decoded} = wire:decode(Msg),

Replacement in test/cable_basic_SUITE.erl at line 61 [5.22]

B:BD[6.1180] → [4.314:352]

    {ok, Decoded} = wire:decode(Bin),

[6.1180]

[4.352]

    {ok, [Msg]} = wire:split_messages(Bin),
    {ok, Decoded} = wire:decode(Msg),

Replacement in test/cable_basic_SUITE.erl at line 72 [5.22]

B:BD[3.562] → [4.397:435]

    {ok, Decoded} = wire:decode(Bin),

[3.562]

[4.435]

    {ok, [Msg]} = wire:split_messages(Bin),
    {ok, Decoded} = wire:decode(Msg),

Deletion in test/cable_basic_SUITE.erl at line 79 [5.22]

B:BD[3.872] → [4.483:484]

B:BD[4.484] → [2.2:642]

∅:D[2.642] → [8.385:386]

∅:D[4.729] → [8.385:386]

∅:D[3.872] → [8.385:386]

B:BD[8.385] → [8.385:386]

B:BD[8.386] → [4.730:1215]


chunk_messages_test() ->
    A = crypto:strong_rand_bytes(5),
    ALen = wire:encode_varint(byte_size(A)),
    B = crypto:strong_rand_bytes(10),
    BLen = wire:encode_varint(byte_size(B)),
    C = crypto:strong_rand_bytes(5),
    CLen = wire:encode_varint(byte_size(C)),
    Payload = <<ALen/bytes, A/bytes, BLen/bytes, B/bytes, CLen/bytes, C/bytes>>,
    {ok, [A, B, C]} = wire:split_messages(Payload).
multiple_in_one_test() ->
     [{binary, Bin1}, _] = examples:hash_response(),
     [{binary, Bin2}, _] = examples:post_request(),
     {multiple, Msgs} = wire:decode(<<Bin1/binary, Bin2/binary>>),
     ?assertEqual(2, length(Msgs)).
%% full circle
encode_channel_state_req_test() ->
    TH = make_test_header(),
    Msg = wire:encode_channel_state_request(TH, "Foo", true),
    {ok, [Header, Body]} = wire:decode(Msg),
    [{requestId, Req}, {circuitId, Circ}, _] = TH,
    ?assertEqual(Req, proplists:get_value(requestId, Header)),
    ?assertEqual(Circ, proplists:get_value(circuitId, Header)),
    ?assertEqual(<<"Foo">>, proplists:get_value(channel, Body)),
    ?assertEqual(1, proplists:get_value(future, Body)).

Deletion in test/cable_basic_SUITE.erl at line 80 [5.22]

B:BD[4.1216] → [4.1216:1364]

make_test_header() ->
    [{requestId, crypto:strong_rand_bytes(4)},
     {circuitId, crypto:strong_rand_bytes(4)},
     {ttl, rand:uniform(20)}].

Replacement in test/cable_basic_SUITE.erl at line 82 [5.22]

B:BD[8.474] → [4.1365:1403]

    {ok, Decoded} = wire:decode(Bin),

[8.474]

[4.1403]

    {ok, [Msg]} = wire:split_messages(Bin),
    {ok, Decoded} = wire:decode(Msg),

Replacement in test/cable_basic_SUITE.erl at line 91 [5.22]

B:BD[8.704] → [4.1449:1487]

    {ok, Decoded} = wire:decode(Bin),

[8.704]

[4.1487]

    {ok, [Msg]} = wire:split_messages(Bin),
    {ok, Decoded} = wire:decode(Msg),

Deletion in test/cable_basic_SUITE.erl at line 97 [5.22]

∅:D[4.1566] → [8.887:888]

B:BD[8.887] → [8.887:888]

B:BD[8.888] → [4.1567:1603]

∅:D[4.1603] → [8.888:1299]

B:BD[8.888] → [8.888:1299]

B:BD[8.1299] → [9.621:622]

∅:D[9.622] → [8.1304:1718]

B:BD[8.1304] → [8.1304:1718]


%% move these to seperate test file
post_text_test() ->
    [{binary, Bin}, {obj, TestObj}] = examples:post_text(),
    Decoded = posts:decode_post_text(Bin),
    [ {pubKey, _PubKey}
    , {links, _Links}
    , {postType, _PostType}
    , {timestamp, _TimeStamp}
    , {channel, Channel}
    , {text, Text}
    ] = Decoded,
    ?assertEqual("default", Channel),
    ?assertEqual(<<"h€llo world"/utf8>>, Text),
    assert_post(Decoded, TestObj).
post_delete_test() ->
    [{binary, Bin}, {obj, TestObj}] = examples:post_delete(),
    Decoded = posts:decode_post_delete(Bin),
    [ {pubKey, _PubKey}
    , {links, _Links}
    , {postType, _PostType}
    , {timestamp, _TimeStamp}
    , {hashes, Hashes}
    ] = Decoded,
    % TODO: assert Hash values
    ?assertEqual(length(maps:get(<<"hashes">>, TestObj)), length(Hashes)),
    assert_post(Decoded, TestObj).

Replacement in test/cable_basic_SUITE.erl at line 98 [5.22]

∅:D[9.624] → [8.1728:2076]

B:BD[8.1728] → [8.1728:2076]

post_info_test() ->
    [{binary, Bin}, {obj, TestObj}] = examples:post_info(),
    Decoded = posts:decode_post_info(Bin),
    [ {pubKey, _PubKey}
    , {links, _Links}
    , {postType, _PostType}
    , {timestamp, _TimeStamp}
    , {infos, Infos}
    ] = Decoded,
    ?assertEqual(#{"name" => "cabler"}, Infos),
    assert_post(Decoded, TestObj).

[9.624]

[8.2076]

%% full circle
encode_channel_state_req_test() ->
    TH = make_test_header(),
    Bin = wire:encode_channel_state_request(TH, "Foo", true),
    {ok, [Msg]} = wire:split_messages(Bin),
    {ok, [Header, Body]} = wire:decode(Msg),
    [{requestId, Req}, {circuitId, Circ}, _] = TH,
    ?assertEqual(Req, proplists:get_value(requestId, Header)),
    ?assertEqual(Circ, proplists:get_value(circuitId, Header)),
    ?assertEqual(<<"Foo">>, proplists:get_value(channel, Body)),
    ?assertEqual(1, proplists:get_value(future, Body)).

Replacement in test/cable_basic_SUITE.erl at line 110 [5.22]

B:BD[8.2077] → [8.2077:2523]

post_topic_test() ->
    [{binary, Bin}, {obj, TestObj}] = examples:post_topic(),
    Decoded = posts:decode_post_topic(Bin),
    [ {pubKey, _PubKey}
    , {links, _Links}
    , {postType, _PostType}
    , {timestamp, _TimeStamp}
    , {channel, Channel}
    , {topic, Topic}
    ] = Decoded,
    ?assertEqual(maps:get(<<"channel">>, TestObj), Channel),
    ?assertEqual(maps:get(<<"topic">>, TestObj), Topic),
    assert_post(Decoded, TestObj).

[8.2077]

[8.2523]

% low-level helper

Replacement in test/cable_basic_SUITE.erl at line 112 [5.22]

B:BD[8.2524] → [8.2524:2889]

post_join_test() ->
    [{binary, Bin}, {obj, TestObj}] = examples:post_join(),
    Decoded = posts:decode_post_join(Bin),
    [ {pubKey, _PubKey}
    , {links, _Links}
    , {postType, _PostType}
    , {timestamp, _TimeStamp}
    , {channel, Channel}
    ] = Decoded,
    ?assertEqual(maps:get(<<"channel">>, TestObj), Channel),
    assert_post(Decoded, TestObj).

[8.2524]

[8.2889]

chunk_messages_test() ->
    A = crypto:strong_rand_bytes(5),
    ALen = wire:encode_varint(byte_size(A)),
    B = crypto:strong_rand_bytes(10),
    BLen = wire:encode_varint(byte_size(B)),
    C = crypto:strong_rand_bytes(5),
    CLen = wire:encode_varint(byte_size(C)),
    Payload = <<ALen/bytes, A/bytes, BLen/bytes, B/bytes, CLen/bytes, C/bytes>>,
    {ok, [A, B, C]} = wire:split_messages(Payload).

Deletion in test/cable_basic_SUITE.erl at line 122 [5.22]

B:BD[8.2890] → [8.2890:3258]

post_leave_test() ->
    [{binary, Bin}, {obj, TestObj}] = examples:post_leave(),
    Decoded = posts:decode_post_leave(Bin),
    [ {pubKey, _PubKey}
    , {links, _Links}
    , {postType, _PostType}
    , {timestamp, _TimeStamp}
    , {channel, Channel}
    ] = Decoded,
    ?assertEqual(maps:get(<<"channel">>, TestObj), Channel),
    assert_post(Decoded, TestObj).

Replacement in test/cable_basic_SUITE.erl at line 126 [5.22]

B:BD[8.3261] → [8.3261:3719]

assert_post(Post, TestObj) ->
    ?assertEqual(maps:get(<<"postType">>, TestObj),  proplists:get_value(postType, Post)),
    ?assertEqual(maps:get(<<"timestamp">>, TestObj), proplists:get_value(timestamp, Post)),
    ?assertEqual(length(maps:get(<<"links">>, TestObj)), length(proplists:get_value(links, Post))),
    PubKey = hex:hexstr_to_bin(binary_to_list(maps:get(<<"publicKey">>, TestObj))),
    ?assertEqual(PubKey, proplists:get_value(pubKey, Post)).

[8.3261]

[6.1381]

make_test_header() ->
    [{requestId, crypto:strong_rand_bytes(4)},
     {circuitId, crypto:strong_rand_bytes(4)},
     {ttl, rand:uniform(20)}].

Deletion in test/cable_basic_SUITE.erl at line 132 [5.22]
∅:D[4.1638] → [8.3720:3808]
B:BD[6.1417] → [8.3720:3808]
```
    ?assertEqual(maps:get(<<"msgLen">>, TestObj), proplists:get_value(msgLen, Header)),
```

Replacement in apps/cable/src/wire.erl at line 51 [5.5336]

∅:D[10.571] → [5.5469:5511]

B:BD[5.5469] → [5.5469:5511]

B:BD[5.5511] → [4.3149:3774]

    {MsgLen, Rest} = decode_varint(Data),
    case MsgLen =:= byte_size(Rest) of
        true ->
            {MsgType, Rest2} = decode_varint(Rest),
            << CircuitId:4/binary, ReqId:4/binary, Rest3/binary >> = Rest2,
            Header = [ {msgLen, MsgLen}
                     , {msgType, MsgType}
                     , {circuitId, CircuitId}
                     , {requestId, ReqId}],
            {ok, [Header, Rest3]};
        false ->
            % we currently assume the passed data is a full message
            % but a client _could_ send multiple in a single (tcp) write
            erlang:error("TODO: implement proper message chunking")
    end.

[10.571]

[5.5775]

    {MsgType, Rest2} = decode_varint(Data),
    << CircuitId:4/binary, ReqId:4/binary, Rest3/binary >> = Rest2,
    Header = [ {msgType, MsgType}
             , {circuitId, CircuitId}
             , {requestId, ReqId}],
    {ok, [Header, Rest3]}.

Replacement in apps/cable/src/peer.erl at line 52 [10.854]

B:BD[10.2319] → [4.4816:5154]

∅:D[4.5154] → [10.2396:2428]

B:BD[10.2396] → [10.2396:2428]

      Result = wire:decode(Data),
      % not sure if it's good to produce this many messages but i'm fine with it for now
      case Result of
          {ok, Msg} ->
              Handler ! {incoming, Conn, Msg};
          {multiple, Msgs} -> lists:foreach(fun(Msg) ->
              Handler ! {incoming, conn, Msg} end, Msgs)
      end,
      decode_loop(State, Conn);

[10.2319]

[10.2428]

      {ok, Msgs} = wire:split_messages(Data),
          lists:foreach(fun(Chunk) ->
                                {ok, Msg} = wire:decode(Chunk),
                                Handler ! {incoming, Conn, Msg}
                        end, Msgs),
          decode_loop(State, Conn);