Skip to content

Commit 9beda57

Browse files
acogoluegnesacogoluegnes
committed
Use "binding key" lingo for super stream creation
Instead of "routing key".
1 parent 809e97f commit 9beda57

File tree

6 files changed

+27
-27
lines changed

6 files changed

+27
-27
lines changed

‎deps/rabbitmq_stream/docs/PROTOCOL.adoc

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -767,13 +767,13 @@ StreamStatsResponse => Key Version CorrelationId ResponseCode Stats
767767
=== CreateSuperStream
768768

769769
```
770-
CreateSuperStream => Key Version CorrelationId Name [Partition] [RoutingKey] Arguments
770+
CreateSuperStream => Key Version CorrelationId Name [Partition] [BindingKey] Arguments
771771
Key => uint16 // 0x001d
772772
Version => uint16
773773
CorrelationId => uint32
774774
Name => string
775775
Partition => string
776-
RoutingKey => string
776+
BindingKey => string
777777
Arguments => [Argument]
778778
Argument => Key Value
779779
Key => string

‎deps/rabbitmq_stream/src/rabbit_stream_manager.erl

Lines changed: 11 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -73,15 +73,15 @@ create_super_stream(VirtualHost,
7373
Name,
7474
Partitions,
7575
Arguments,
76-
RoutingKeys,
76+
BindingKeys,
7777
Username) ->
7878
gen_server:call(?MODULE,
7979
{create_super_stream,
8080
VirtualHost,
8181
Name,
8282
Partitions,
8383
Arguments,
84-
RoutingKeys,
84+
BindingKeys,
8585
Username}).
8686

8787
-spec delete_super_stream(binary(), binary(), binary()) ->
@@ -226,7 +226,7 @@ handle_call({create_super_stream,
226226
Name,
227227
Partitions,
228228
Arguments,
229-
RoutingKeys,
229+
BindingKeys,
230230
Username},
231231
_From, State) ->
232232
case validate_super_stream_creation(VirtualHost, Name, Partitions) of
@@ -273,7 +273,7 @@ handle_call({create_super_stream,
273273
add_super_stream_bindings(VirtualHost,
274274
Name,
275275
Partitions,
276-
RoutingKeys,
276+
BindingKeys,
277277
Username),
278278
case BindingsResult of
279279
ok ->
@@ -758,15 +758,15 @@ declare_super_stream_exchange(VirtualHost, Name, Username) ->
758758
add_super_stream_bindings(VirtualHost,
759759
Name,
760760
Partitions,
761-
RoutingKeys,
761+
BindingKeys,
762762
Username) ->
763-
PartitionsRoutingKeys = lists:zip(Partitions, RoutingKeys),
763+
PartitionsBindingKeys = lists:zip(Partitions, BindingKeys),
764764
BindingsResult =
765-
lists:foldl(fun ({Partition, RoutingKey}, {ok, Order}) ->
765+
lists:foldl(fun ({Partition, BindingKey}, {ok, Order}) ->
766766
case add_super_stream_binding(VirtualHost,
767767
Name,
768768
Partition,
769-
RoutingKey,
769+
BindingKey,
770770
Order,
771771
Username)
772772
of
@@ -778,7 +778,7 @@ add_super_stream_bindings(VirtualHost,
778778
(_, {{error, _Reason}, _Order} = Acc) ->
779779
Acc
780780
end,
781-
{ok, 0}, PartitionsRoutingKeys),
781+
{ok, 0}, PartitionsBindingKeys),
782782
case BindingsResult of
783783
{ok, _} ->
784784
ok;
@@ -789,7 +789,7 @@ add_super_stream_bindings(VirtualHost,
789789
add_super_stream_binding(VirtualHost,
790790
SuperStream,
791791
Partition,
792-
RoutingKey,
792+
BindingKey,
793793
Order,
794794
Username) ->
795795
{ok, ExchangeNameBin} =
@@ -806,7 +806,7 @@ add_super_stream_binding(VirtualHost,
806806
Order),
807807
case rabbit_binding:add(#binding{source = ExchangeName,
808808
destination = QueueName,
809-
key = RoutingKey,
809+
key = BindingKey,
810810
args = Arguments},
811811
fun (_X, Q) when ?is_amqqueue(Q) ->
812812
try

‎deps/rabbitmq_stream/src/rabbit_stream_reader.erl

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -2883,7 +2883,7 @@ handle_frame_post_auth(Transport,
28832883
user = #user{username = Username} = User} = Connection,
28842884
State,
28852885
{request, CorrelationId,
2886-
{create_super_stream, SuperStream, Partitions, RoutingKeys, Arguments}}) ->
2886+
{create_super_stream, SuperStream, Partitions, BindingKeys, Arguments}}) ->
28872887
case rabbit_stream_utils:enforce_correct_name(SuperStream) of
28882888
{ok, SuperStreamName} ->
28892889
case rabbit_stream_utils:check_super_stream_management_permitted(VirtualHost,
@@ -2895,7 +2895,7 @@ handle_frame_post_auth(Transport,
28952895
SuperStreamName,
28962896
Partitions,
28972897
Arguments,
2898-
RoutingKeys,
2898+
BindingKeys,
28992899
Username) of
29002900
ok ->
29012901
rabbit_log:debug("Created super stream ~tp", [SuperStreamName]),

‎deps/rabbitmq_stream/test/rabbit_stream_SUITE.erl

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -321,8 +321,8 @@ test_super_stream_creation_deletion(Config) ->
321321

322322
Ss = atom_to_binary(?FUNCTION_NAME, utf8),
323323
Partitions = [unicode:characters_to_binary([Ss, <<"-">>, integer_to_binary(N)]) || N <- lists:seq(0, 2)],
324-
Rks = [integer_to_binary(N) || N <- lists:seq(0, 2)],
325-
SsCreationFrame = frame({request, 1, {create_super_stream, Ss, Partitions, Rks, #{}}}),
324+
Bks = [integer_to_binary(N) || N <- lists:seq(0, 2)],
325+
SsCreationFrame = frame({request, 1, {create_super_stream, Ss, Partitions, Bks, #{}}}),
326326
ok = T:send(S, SsCreationFrame),
327327
{Cmd1, _} = receive_commands(T, S, C),
328328
?assertMatch({response, 1, {create_super_stream, ?RESPONSE_CODE_OK}},
@@ -340,7 +340,7 @@ test_super_stream_creation_deletion(Config) ->
340340
?assertMatch({response, 1, {route, ?RESPONSE_CODE_OK, _}}, Command),
341341
{response, 1, {route, ?RESPONSE_CODE_OK, [P]}} = Command,
342342
?assertEqual(unicode:characters_to_binary([Ss, <<"-">>, Rk]), P)
343-
end || Rk <- Rks],
343+
end || Rk <- Bks],
344344

345345
SsDeletionFrame = frame({request, 1, {delete_super_stream, Ss}}),
346346
ok = T:send(S, SsDeletionFrame),

‎deps/rabbitmq_stream_common/src/rabbit_stream_core.erl

Lines changed: 7 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -116,7 +116,7 @@
116116
[{Command :: atom(), MinVersion :: command_version(),
117117
MaxVersion :: command_version()}]} |
118118
{stream_stats, Stream :: binary()} |
119-
{create_super_stream, stream_name(), Partitions :: [binary()], RoutingKeys :: [binary()], Args :: #{binary() => binary()}} |
119+
{create_super_stream, stream_name(), Partitions :: [binary()], BindingKeys :: [binary()], Args :: #{binary() => binary()}} |
120120
{delete_super_stream, stream_name()}} |
121121
{response, correlation_id(),
122122
{declare_publisher |
@@ -608,12 +608,12 @@ request_body({exchange_command_versions = Tag, CommandVersions}) ->
608608
{Tag, [<<CommandVersionsLength:32>>, CommandVersionsBin]};
609609
request_body({stream_stats = Tag, Stream}) ->
610610
{Tag, <<?STRING(Stream)>>};
611-
request_body({create_super_stream = Tag, SuperStream, Partitions, RoutingKeys, Args}) ->
611+
request_body({create_super_stream = Tag, SuperStream, Partitions, BindingKeys, Args}) ->
612612
PartitionsBin = generate_list(Partitions),
613-
RoutingKeysBin = generate_list(RoutingKeys),
613+
BindingKeysBin = generate_list(BindingKeys),
614614
ArgsBin = generate_map(Args),
615615
{Tag, [<<?STRING(SuperStream), (length(Partitions)):32>>, PartitionsBin,
616-
<<(length(RoutingKeys)):32>>, RoutingKeysBin,
616+
<<(length(BindingKeys)):32>>, BindingKeysBin,
617617
<<(map_size(Args)):32>>, ArgsBin]};
618618
request_body({delete_super_stream = Tag, SuperStream}) ->
619619
{Tag, <<?STRING(SuperStream)>>}.
@@ -903,10 +903,10 @@ parse_request(<<?REQUEST:1,
903903
?STRING(StreamSize, Stream),
904904
PartitionsCount:32,
905905
Rest0/binary>>) ->
906-
{Partitions, <<RoutingKeysCount:32, Rest1/binary>>} = list_of_strings(PartitionsCount, Rest0),
907-
{RoutingKeys, <<_ArgumentsCount:32, Rest2/binary>>} = list_of_strings(RoutingKeysCount, Rest1),
906+
{Partitions, <<BindingKeysCount:32, Rest1/binary>>} = list_of_strings(PartitionsCount, Rest0),
907+
{BindingKeys, <<_ArgumentsCount:32, Rest2/binary>>} = list_of_strings(BindingKeysCount, Rest1),
908908
Args = parse_map(Rest2, #{}),
909-
request(CorrelationId, {create_super_stream, Stream, Partitions, RoutingKeys, Args});
909+
request(CorrelationId, {create_super_stream, Stream, Partitions, BindingKeys, Args});
910910
parse_request(<<?REQUEST:1,
911911
?COMMAND_DELETE_SUPER_STREAM:15,
912912
?VERSION_1:16,

‎deps/rabbitmq_stream_common/test/rabbit_stream_core_SUITE.erl

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -112,11 +112,11 @@ roundtrip(_Config) ->
112112
test_roundtrip({request, 99, {stream_stats, <<"stream_name">>}}),
113113
test_roundtrip({request, 99,
114114
{create_super_stream, <<"hello">>,
115-
[<<"stream1">>, <<"stream2">>, <<"stream3">>], [<<"rk1">>, <<"rk2">>, <<"rk3">>],
115+
[<<"stream1">>, <<"stream2">>, <<"stream3">>], [<<"bk1">>, <<"bk2">>, <<"bk3">>],
116116
Args}}),
117117
test_roundtrip({request, 99,
118118
{create_super_stream, <<"super_stream_name">>,
119-
[<<"stream1">>, <<"stream2">>, <<"stream3">>], [<<"rk1">>, <<"rk2">>, <<"rk3">>],
119+
[<<"stream1">>, <<"stream2">>, <<"stream3">>], [<<"bk1">>, <<"bk2">>, <<"bk3">>],
120120
#{}}}),
121121
test_roundtrip({request, 99, {delete_super_stream, <<"super_stream_name">>}}),
122122

0 commit comments

Comments
 (0)