%% This Source Code Form is subject to the terms of the Mozilla Public %% License, v. 2.0. If a copy of the MPL was not distributed with this %% file, You can obtain one at https://mozilla.org/MPL/2.0/. %% %% Copyright (c) 2007-2024 Broadcom. All Rights Reserved. The term “Broadcom” refers to Broadcom Inc. and/or its subsidiaries. All rights reserved. %% -module(amqp10_msg). -export([from_amqp_records/1, to_amqp_records/1, % "read" api delivery_id/1, delivery_tag/1, handle/1, settled/1, message_format/1, headers/1, header/2, delivery_annotations/1, message_annotations/1, properties/1, application_properties/1, body/1, body_bin/1, footer/1, % "write" api new/2, new/3, set_handle/2, set_settled/2, set_message_format/2, set_headers/2, set_properties/2, set_application_properties/2, set_delivery_annotations/2, set_message_annotations/2 ]). -include_lib("amqp10_common/include/amqp10_framing.hrl"). -type maybe(T) :: T | undefined. -type delivery_tag() :: binary(). -type content_type() :: term(). % TODO: refine -type content_encoding() :: term(). % TODO: refine % annotations keys are restricted to be of type symbol or of type ulong -type annotations_key() :: binary() | non_neg_integer(). -type header_key() :: durable | priority | ttl | first_acquirer | delivery_count. -type amqp10_header() :: #{durable => boolean(), % false priority => byte(), % 4 ttl => maybe(non_neg_integer()), first_acquirer => boolean(), % false delivery_count => non_neg_integer()}. % 0 -type amqp10_properties() :: #{message_id => maybe(any()), user_id => maybe(binary()), to => maybe(any()), subject => maybe(binary()), reply_to => maybe(any()), correlation_id => maybe(any()), content_type => maybe(content_type()), content_encoding => maybe(content_encoding()), absolute_expiry_time => maybe(non_neg_integer()), creation_time => maybe(non_neg_integer()), group_id => maybe(binary()), group_sequence => maybe(non_neg_integer()), reply_to_group_id => maybe(binary())}. -type amqp10_body() :: [#'v1_0.data'{}] | [#'v1_0.amqp_sequence'{}] | #'v1_0.amqp_value'{}. -record(amqp10_msg, {transfer :: #'v1_0.transfer'{}, header :: maybe(#'v1_0.header'{}), delivery_annotations :: maybe(#'v1_0.delivery_annotations'{}), message_annotations :: maybe(#'v1_0.message_annotations'{}), properties :: maybe(#'v1_0.properties'{}), application_properties :: maybe(#'v1_0.application_properties'{}), body :: amqp10_body() | unset, footer :: maybe(#'v1_0.footer'{}) }). -opaque amqp10_msg() :: #amqp10_msg{}. -export_type([amqp10_msg/0, amqp10_header/0, amqp10_properties/0, amqp10_body/0, delivery_tag/0, annotations_key/0 ]). -define(record_to_tuplelist(Rec, Ref), lists:zip(record_info(fields, Rec), tl(tuple_to_list(Ref)))). %% API functions -spec from_amqp_records([amqp10_client_types:amqp10_msg_record()]) -> amqp10_msg(). from_amqp_records([#'v1_0.transfer'{} = Transfer | Records]) -> lists:foldl(fun parse_from_amqp/2, #amqp10_msg{transfer = Transfer, body = unset}, Records). -spec to_amqp_records(amqp10_msg()) -> [amqp10_client_types:amqp10_msg_record()]. to_amqp_records(#amqp10_msg{transfer = T, header = H, delivery_annotations = DAs, message_annotations = MAs, properties = Ps, application_properties = APs, body = B, footer = F }) -> L = lists:flatten([T, H, DAs, MAs, Ps, APs, B, F]), lists:filter(fun has_value/1, L). -spec delivery_tag(amqp10_msg()) -> delivery_tag(). delivery_tag(#amqp10_msg{transfer = #'v1_0.transfer'{delivery_tag = Tag}}) -> unpack(Tag). -spec delivery_id(amqp10_msg()) -> non_neg_integer(). delivery_id(#amqp10_msg{transfer = #'v1_0.transfer'{delivery_id = Id}}) -> unpack(Id). -spec handle(amqp10_msg()) -> non_neg_integer(). handle(#amqp10_msg{transfer = #'v1_0.transfer'{handle = Handle}}) -> unpack(Handle). -spec settled(amqp10_msg()) -> boolean(). settled(#amqp10_msg{transfer = #'v1_0.transfer'{settled = Settled}}) -> Settled. % First 3 octets are the format % the last 1 octet is the version % See 2.8.11 in the spec -spec message_format(amqp10_msg()) -> maybe({non_neg_integer(), non_neg_integer()}). message_format(#amqp10_msg{transfer = #'v1_0.transfer'{message_format = undefined}}) -> undefined; message_format(#amqp10_msg{transfer = #'v1_0.transfer'{message_format = {uint, MF}}}) -> <> = <>, {Format, Version}. -spec headers(amqp10_msg()) -> amqp10_header(). headers(#amqp10_msg{header = undefined}) -> #{}; headers(#amqp10_msg{header = #'v1_0.header'{durable = Durable, priority = Priority, ttl = Ttl, first_acquirer = FA, delivery_count = DC}}) -> Fields = [{durable, header_value(durable, Durable)}, {priority, header_value(priority, Priority)}, {ttl, header_value(ttl, Ttl)}, {first_acquirer, header_value(first_acquirer, FA)}, {delivery_count, header_value(delivery_count, DC)}], lists:foldl(fun ({_Key, undefined}, Acc) -> Acc; ({Key, Value}, Acc) -> Acc#{Key => Value} end, #{}, Fields). -spec header(header_key(), amqp10_msg()) -> term(). header(durable = K, #amqp10_msg{header = #'v1_0.header'{durable = D}}) -> header_value(K, D); header(priority = K, #amqp10_msg{header = #'v1_0.header'{priority = D}}) -> header_value(K, D); header(ttl = K, #amqp10_msg{header = #'v1_0.header'{ttl = D}}) -> header_value(K, D); header(first_acquirer = K, #amqp10_msg{header = #'v1_0.header'{first_acquirer = D}}) -> header_value(K, D); header(delivery_count = K, #amqp10_msg{header = #'v1_0.header'{delivery_count = D}}) -> header_value(K, D); header(K, #amqp10_msg{header = undefined}) -> header_value(K, undefined). -spec delivery_annotations(amqp10_msg()) -> #{annotations_key() => any()}. delivery_annotations(#amqp10_msg{delivery_annotations = undefined}) -> #{}; delivery_annotations(#amqp10_msg{delivery_annotations = #'v1_0.delivery_annotations'{content = DAs}}) -> lists:foldl(fun({K, V}, Acc) -> Acc#{unpack(K) => unpack(V)} end, #{}, DAs). -spec message_annotations(amqp10_msg()) -> #{annotations_key() => any()}. message_annotations(#amqp10_msg{message_annotations = undefined}) -> #{}; message_annotations(#amqp10_msg{message_annotations = #'v1_0.message_annotations'{content = MAs}}) -> lists:foldl(fun({K, V}, Acc) -> Acc#{unpack(K) => unpack(V)} end, #{}, MAs). -spec properties(amqp10_msg()) -> amqp10_properties(). properties(#amqp10_msg{properties = undefined}) -> #{}; properties(#amqp10_msg{properties = Props}) -> Fields = ?record_to_tuplelist('v1_0.properties', Props), lists:foldl(fun ({_Key, undefined}, Acc) -> Acc; ({Key, Value}, Acc) -> Acc#{Key => unpack(Value)} end, #{}, Fields). % application property values can be simple types - no maps or lists -spec application_properties(amqp10_msg()) -> #{binary() => binary() | integer() | string()}. application_properties(#amqp10_msg{application_properties = undefined}) -> #{}; application_properties( #amqp10_msg{application_properties = #'v1_0.application_properties'{content = MAs}}) -> lists:foldl(fun({K, V}, Acc) -> Acc#{unpack(K) => unpack(V)} end, #{}, MAs). -spec footer(amqp10_msg()) -> #{annotations_key() => any()}. footer(#amqp10_msg{footer = undefined}) -> #{}; footer(#amqp10_msg{footer = #'v1_0.footer'{content = Footer}}) -> lists:foldl(fun({K, V}, Acc) -> Acc#{unpack(K) => unpack(V)} end, #{}, Footer). -spec body(amqp10_msg()) -> [binary()] | [#'v1_0.amqp_sequence'{}] | #'v1_0.amqp_value'{}. body(#amqp10_msg{body = [#'v1_0.data'{} | _] = Data}) -> [Content || #'v1_0.data'{content = Content} <- Data]; body(#amqp10_msg{body = Body}) -> Body. %% @doc Returns the binary representation -spec body_bin(amqp10_msg()) -> binary(). body_bin(#amqp10_msg{body = [#'v1_0.data'{content = Bin}]}) when is_binary(Bin) -> Bin; body_bin(#amqp10_msg{body = Data}) when is_list(Data) -> iolist_to_binary([amqp10_framing:encode_bin(D) || D <- Data]); body_bin(#amqp10_msg{body = #'v1_0.amqp_value'{} = Body}) -> %% TODO: to avoid unnecessary decoding and re-encoding we could amend %% the parse to provide the body in a lazy fashion, only decoding when %% reading. For now we just re-encode it. iolist_to_binary(amqp10_framing:encode_bin(Body)). %% @doc Create a new amqp10 message using the specified delivery tag, body %% and settlement state. Settled=true means the message is considered settled %% as soon as sent and no disposition will be issued by the receiver. %% Settled=false will delay settlement until a disposition has been received. %% A disposition will be notified to the sender by a message of the %% following stucture: %% {amqp10_disposition, {accepted | rejected, DeliveryTag}} -spec new(delivery_tag(), amqp10_body() | binary(), boolean()) -> amqp10_msg(). new(DeliveryTag, Body, Settled) when is_binary(Body) -> #amqp10_msg{transfer = #'v1_0.transfer'{delivery_tag = {binary, DeliveryTag}, settled = Settled, message_format = {uint, 0}}, body = [#'v1_0.data'{content = Body}]}; new(DeliveryTag, Body, Settled) -> % TODO: constrain to amqp types #amqp10_msg{transfer = #'v1_0.transfer'{delivery_tag = {binary, DeliveryTag}, settled = Settled, message_format = {uint, 0}}, body = Body}. %% @doc Create a new settled amqp10 message using the specified delivery tag %% and body. -spec new(delivery_tag(), amqp10_body() | binary()) -> amqp10_msg(). new(DeliveryTag, Body) -> new(DeliveryTag, Body, false). % First 3 octets are the format % the last 1 octet is the version % See 2.8.11 in the spec %% @doc Set the message format. -spec set_message_format({non_neg_integer(), non_neg_integer()}, amqp10_msg()) -> amqp10_msg(). set_message_format({Format, Version}, #amqp10_msg{transfer = T} = Msg) -> <> = <>, Msg#amqp10_msg{transfer = T#'v1_0.transfer'{message_format = {uint, MsgFormat}}}. %% @doc Set the link handle used for the message transfer. -spec set_handle(non_neg_integer(), amqp10_msg()) -> amqp10_msg(). set_handle(Handle, #amqp10_msg{transfer = T} = Msg) -> Msg#amqp10_msg{transfer = T#'v1_0.transfer'{handle = {uint, Handle}}}. %% @doc Set the settledment mode. %% Settled=true means the message is considered settled %% as soon as sent and no disposition will be issued by the receiver. %% Settled=false will delay settlement until a disposition has been received. %% A disposition will be notified to the sender by a message of the %% following stucture: %% {amqp10_disposition, {accepted | rejected, DeliveryTag}} -spec set_settled(boolean(), amqp10_msg()) -> amqp10_msg(). set_settled(Settled, #amqp10_msg{transfer = T} = Msg) -> Msg#amqp10_msg{transfer = T#'v1_0.transfer'{settled = Settled}}. %% @doc Set amqp message headers. -spec set_headers(#{atom() => any()}, amqp10_msg()) -> amqp10_msg(). set_headers(Headers, #amqp10_msg{header = undefined} = Msg) -> set_headers(Headers, Msg#amqp10_msg{header = #'v1_0.header'{}}); set_headers(Headers, #amqp10_msg{header = Current} = Msg) -> H = maps:fold(fun(durable, V, Acc) -> Acc#'v1_0.header'{durable = V}; (priority, V, Acc) -> Acc#'v1_0.header'{priority = {ubyte, V}}; (first_acquirer, V, Acc) -> Acc#'v1_0.header'{first_acquirer = V}; (ttl, V, Acc) -> Acc#'v1_0.header'{ttl = {uint, V}}; (delivery_count, V, Acc) -> Acc#'v1_0.header'{delivery_count = {uint, V}} end, Current, Headers), Msg#amqp10_msg{header = H}. %% @doc Set amqp message properties. -spec set_properties(amqp10_properties(), amqp10_msg()) -> amqp10_msg(). set_properties(Props, #amqp10_msg{properties = undefined} = Msg) -> set_properties(Props, Msg#amqp10_msg{properties = #'v1_0.properties'{}}); set_properties(Props, #amqp10_msg{properties = Current} = Msg) -> % TODO many fields are `any` types and we need to try to type tag them P = maps:fold(fun(message_id, V, Acc) when is_binary(V) -> % message_id can be any type but we restrict it here Acc#'v1_0.properties'{message_id = utf8(V)}; (user_id, V, Acc) when is_binary(V) orelse is_list(V) -> Acc#'v1_0.properties'{user_id = binary(V)}; (to, V, Acc) -> Acc#'v1_0.properties'{to = utf8(V)}; (subject, V, Acc) -> Acc#'v1_0.properties'{subject = utf8(V)}; (reply_to, V, Acc) -> Acc#'v1_0.properties'{reply_to = utf8(V)}; (correlation_id, V, Acc) -> Acc#'v1_0.properties'{correlation_id = utf8(V)}; (content_type, V, Acc) -> Acc#'v1_0.properties'{content_type = sym(V)}; (content_encoding, V, Acc) -> Acc#'v1_0.properties'{content_encoding = sym(V)}; (absolute_expiry_time, V, Acc) -> Acc#'v1_0.properties'{absolute_expiry_time = {timestamp, V}}; (creation_time, V, Acc) -> Acc#'v1_0.properties'{creation_time = {timestamp, V}}; (group_id, V, Acc) -> Acc#'v1_0.properties'{group_id = utf8(V)}; (group_sequence, V, Acc) -> Acc#'v1_0.properties'{group_sequence = uint(V)}; (reply_to_group_id, V, Acc) -> Acc#'v1_0.properties'{reply_to_group_id = utf8(V)} end, Current, Props), Msg#amqp10_msg{properties = P}. -spec set_application_properties(#{binary() | string() => binary() | integer() | string()}, amqp10_msg()) -> amqp10_msg(). set_application_properties(Props, #amqp10_msg{application_properties = undefined} = Msg) -> APs = #'v1_0.application_properties'{content = []}, set_application_properties(Props, Msg#amqp10_msg{application_properties = APs}); set_application_properties( Props0, #amqp10_msg{application_properties = #'v1_0.application_properties'{content = APs0}} = Msg) -> Props = maps:fold(fun (K, V, S) -> S#{utf8(K) => wrap_ap_value(V)} end, maps:from_list(APs0), Props0), APs = #'v1_0.application_properties'{content = maps:to_list(Props)}, Msg#amqp10_msg{application_properties = APs}. -spec set_delivery_annotations(#{binary() => binary() | integer() | string()}, amqp10_msg()) -> amqp10_msg(). set_delivery_annotations(Props, #amqp10_msg{delivery_annotations = undefined} = Msg) -> Anns = #'v1_0.delivery_annotations'{content = []}, set_delivery_annotations(Props, Msg#amqp10_msg{delivery_annotations = Anns}); set_delivery_annotations( Props0, #amqp10_msg{delivery_annotations = #'v1_0.delivery_annotations'{content = Anns0}} = Msg) -> Anns = maps:fold(fun (K, V, S) -> S#{sym(K) => wrap_ap_value(V)} end, maps:from_list(Anns0), Props0), Anns1 = #'v1_0.delivery_annotations'{content = maps:to_list(Anns)}, Msg#amqp10_msg{delivery_annotations = Anns1}. -spec set_message_annotations(#{binary() => binary() | integer() | string()}, amqp10_msg()) -> amqp10_msg(). set_message_annotations(Props, #amqp10_msg{message_annotations = undefined} = Msg) -> Anns = #'v1_0.message_annotations'{content = []}, set_message_annotations(Props, Msg#amqp10_msg{message_annotations = Anns}); set_message_annotations( Props0, #amqp10_msg{message_annotations = #'v1_0.message_annotations'{content = Anns0}} = Msg) -> Anns = maps:fold(fun (K, V, S) -> S#{sym(K) => wrap_ap_value(V)} end, maps:from_list(Anns0), Props0), Anns1 = #'v1_0.message_annotations'{content = maps:to_list(Anns)}, Msg#amqp10_msg{message_annotations = Anns1}. wrap_ap_value(true) -> {boolean, true}; wrap_ap_value(false) -> {boolean, false}; wrap_ap_value(V) when is_integer(V) andalso V >= 0 -> {uint, V}; wrap_ap_value(V) when is_integer(V) andalso V < 0 -> {int, V}; wrap_ap_value(F) when is_float(F) -> {double, F}; wrap_ap_value(V) when is_binary(V) -> utf8(V); wrap_ap_value(V) when is_list(V) -> utf8(list_to_binary(V)); wrap_ap_value(V) when is_atom(V) -> utf8(atom_to_list(V)). %% LOCAL header_value(durable, undefined) -> false; header_value(priority, undefined) -> 4; header_value(first_acquirer, undefined) -> false; header_value(delivery_count, undefined) -> 0; header_value(Key, {_Type, Value}) -> header_value(Key, Value); header_value(_Key, Value) -> Value. parse_from_amqp(#'v1_0.header'{} = Header, AmqpMsg) -> AmqpMsg#amqp10_msg{header = Header}; parse_from_amqp(#'v1_0.delivery_annotations'{} = DAS, AmqpMsg) -> AmqpMsg#amqp10_msg{delivery_annotations = DAS}; parse_from_amqp(#'v1_0.message_annotations'{} = DAS, AmqpMsg) -> AmqpMsg#amqp10_msg{message_annotations = DAS}; parse_from_amqp(#'v1_0.properties'{} = Header, AmqpMsg) -> AmqpMsg#amqp10_msg{properties = Header}; parse_from_amqp(#'v1_0.application_properties'{} = APs, AmqpMsg) -> AmqpMsg#amqp10_msg{application_properties = APs}; parse_from_amqp(#'v1_0.amqp_value'{} = Value, AmqpMsg) -> AmqpMsg#amqp10_msg{body = Value}; parse_from_amqp(#'v1_0.amqp_sequence'{} = Seq, AmqpMsg) -> AmqpMsg#amqp10_msg{body = [Seq]}; parse_from_amqp(#'v1_0.data'{} = Data, AmqpMsg) -> AmqpMsg#amqp10_msg{body = [Data]}; parse_from_amqp(#'v1_0.footer'{} = Header, AmqpMsg) -> AmqpMsg#amqp10_msg{footer = Header}. unpack(V) -> amqp10_client_types:unpack(V). utf8(V) -> amqp10_client_types:utf8(V). sym(B) when is_list(B) -> {symbol, list_to_binary(B)}; sym(B) when is_binary(B) -> {symbol, B}. uint(B) -> {uint, B}. binary(B) when is_binary(B) -> {binary, B}; binary(B) when is_list(B) -> {binary, erlang:list_to_binary(B)}. has_value(undefined) -> false; has_value(_) -> true.