|
9 | 9 | -behaviour(gen_statem). |
10 | 10 |
|
11 | 11 | -include("amqp10_client.hrl"). |
| 12 | +-include("amqp10_client_internal.hrl"). |
12 | 13 | -include_lib("amqp10_common/include/amqp10_framing.hrl"). |
13 | 14 | -include_lib("amqp10_common/include/amqp10_types.hrl"). |
14 | 15 |
|
|
86 | 87 | -type attach_role() :: {sender, target_def()} | {receiver, source_def(), pid()}. |
87 | 88 |
|
88 | 89 | % http://www.amqp.org/specification/1.0/filters |
89 | | --type filter() :: #{binary() => binary() | map() | list(binary())}. |
| 90 | +-type filter() :: #{binary() => #filter{} | binary() | map() | list(binary())}. |
90 | 91 | -type max_message_size() :: undefined | non_neg_integer(). |
91 | 92 | -type footer_opt() :: crc32 | adler32. |
92 | 93 |
|
@@ -781,29 +782,39 @@ translate_filters(Filters) |
781 | 782 | when map_size(Filters) =:= 0 -> |
782 | 783 | undefined; |
783 | 784 | translate_filters(Filters) -> |
784 | | - {map, |
785 | | - maps:fold( |
786 | | - fun |
787 | | - (<<"apache.org:legacy-amqp-headers-binding:map">> = K, V, Acc) when is_map(V) -> |
788 | | - %% special case conversion |
789 | | - Key = sym(K), |
790 | | - [{Key, {described, Key, translate_legacy_amqp_headers_binding(V)}} | Acc]; |
791 | | - (K, V, Acc) when is_binary(K) -> |
792 | | - %% try treat any filter value generically |
793 | | - Key = sym(K), |
794 | | - Value = filter_value_type(V), |
795 | | - [{Key, {described, Key, Value}} | Acc] |
796 | | - end, [], Filters)}. |
797 | | - |
798 | | -filter_value_type(V) when is_binary(V) -> |
| 785 | + {map, lists:map( |
| 786 | + fun({Name, #filter{descriptor = Desc, |
| 787 | + value = V}}) |
| 788 | + when is_binary(Name) -> |
| 789 | + Descriptor = if is_binary(Desc) -> {symbol, Desc}; |
| 790 | + is_integer(Desc) -> {ulong, Desc} |
| 791 | + end, |
| 792 | + {{symbol, Name}, {described, Descriptor, V}}; |
| 793 | + ({<<"apache.org:legacy-amqp-headers-binding:map">> = K, V}) |
| 794 | + when is_map(V) -> |
| 795 | + %% special case conversion |
| 796 | + Key = sym(K), |
| 797 | + Val = translate_legacy_amqp_headers_binding(V), |
| 798 | + {Key, {described, Key, Val}}; |
| 799 | + ({K, V}) |
| 800 | + when is_binary(K) -> |
| 801 | + Key = {symbol, K}, |
| 802 | + Val = filter_value_type(V), |
| 803 | + {Key, {described, Key, Val}} |
| 804 | + end, maps:to_list(Filters))}. |
| 805 | + |
| 806 | +filter_value_type(V) |
| 807 | + when is_binary(V) -> |
799 | 808 | %% this is clearly not always correct |
800 | 809 | {utf8, V}; |
801 | 810 | filter_value_type(V) |
802 | 811 | when is_integer(V) andalso V >= 0 -> |
803 | 812 | {uint, V}; |
804 | | -filter_value_type(VList) when is_list(VList) -> |
| 813 | +filter_value_type(VList) |
| 814 | + when is_list(VList) -> |
805 | 815 | {list, [filter_value_type(V) || V <- VList]}; |
806 | | -filter_value_type({T, _} = V) when is_atom(T) -> |
| 816 | +filter_value_type({T, _} = V) |
| 817 | + when is_atom(T) -> |
807 | 818 | %% looks like an already tagged type, just pass it through |
808 | 819 | V. |
809 | 820 |
|
@@ -1507,16 +1518,17 @@ translate_filters_selector_filter_test() -> |
1507 | 1518 | } = translate_filters(#{<<"apache.org:selector-filter:string">> => <<"amqp.annotation.x-opt-enqueuedtimeutc > 123456789">>}). |
1508 | 1519 |
|
1509 | 1520 | translate_filters_multiple_filters_test() -> |
1510 | | - {map, |
1511 | | - [ |
1512 | | - {{symbol, <<"apache.org:selector-filter:string">>}, |
1513 | | - {described, {symbol, <<"apache.org:selector-filter:string">>}, |
1514 | | - {utf8, <<"amqp.annotation.x-opt-enqueuedtimeutc > 123456789">>}}}, |
1515 | | - {{symbol, <<"apache.org:legacy-amqp-direct-binding:string">>}, |
1516 | | - {described, {symbol, <<"apache.org:legacy-amqp-direct-binding:string">>}, {utf8,<<"my topic">>}}} |
1517 | | - ] |
1518 | | - } = translate_filters(#{ |
1519 | | - <<"apache.org:legacy-amqp-direct-binding:string">> => <<"my topic">>, |
1520 | | - <<"apache.org:selector-filter:string">> => <<"amqp.annotation.x-opt-enqueuedtimeutc > 123456789">> |
1521 | | - }). |
| 1521 | + {map, Actual} = translate_filters( |
| 1522 | + #{ |
| 1523 | + <<"apache.org:legacy-amqp-direct-binding:string">> => <<"my topic">>, |
| 1524 | + <<"apache.org:selector-filter:string">> => <<"amqp.annotation.x-opt-enqueuedtimeutc > 123456789">> |
| 1525 | + }), |
| 1526 | + Expected = [{{symbol, <<"apache.org:selector-filter:string">>}, |
| 1527 | + {described, {symbol, <<"apache.org:selector-filter:string">>}, |
| 1528 | + {utf8, <<"amqp.annotation.x-opt-enqueuedtimeutc > 123456789">>}}}, |
| 1529 | + {{symbol, <<"apache.org:legacy-amqp-direct-binding:string">>}, |
| 1530 | + {described, {symbol, <<"apache.org:legacy-amqp-direct-binding:string">>}, {utf8,<<"my topic">>}}}], |
| 1531 | + ActualSorted = lists:sort(Actual), |
| 1532 | + ExpectedSorted = lists:sort(Expected), |
| 1533 | + ExpectedSorted = ActualSorted. |
1522 | 1534 | -endif. |
0 commit comments