|
| 1 | +%% This Source Code Form is subject to the terms of the Mozilla Public |
| 2 | +%% License, v. 2.0. If a copy of the MPL was not distributed with this |
| 3 | +%% file, You can obtain one at https://mozilla.org/MPL/2.0/. |
| 4 | +%% |
| 5 | +%% Copyright (c) 2007-2025 Broadcom. All Rights Reserved. The term “Broadcom” refers to Broadcom Inc. and/or its subsidiaries. All rights reserved. |
| 6 | +%% |
| 7 | + |
| 8 | +-module(rabbit_stream_super_stream_mgmt). |
| 9 | + |
| 10 | +-behaviour(rabbit_mgmt_extension). |
| 11 | + |
| 12 | +-export([dispatcher/0, |
| 13 | + web_ui/0]). |
| 14 | +-export([init/2, |
| 15 | + content_types_accepted/2, |
| 16 | + is_authorized/2, |
| 17 | + resource_exists/2, |
| 18 | + allowed_methods/2, |
| 19 | + accept_content/2]). |
| 20 | +-export([variances/2]). |
| 21 | + |
| 22 | +-include_lib("rabbitmq_management_agent/include/rabbit_mgmt_records.hrl"). |
| 23 | +-include_lib("rabbit_common/include/rabbit.hrl"). |
| 24 | + |
| 25 | +-define(DEFAULT_RPC_TIMEOUT, 30_000). |
| 26 | + |
| 27 | +dispatcher() -> |
| 28 | + [{"/stream/super-streams/:vhost/:name", ?MODULE, []}]. |
| 29 | + |
| 30 | +web_ui() -> |
| 31 | + []. |
| 32 | + |
| 33 | +%%-------------------------------------------------------------------- |
| 34 | + |
| 35 | +init(Req, _State) -> |
| 36 | + {cowboy_rest, |
| 37 | + rabbit_mgmt_headers:set_common_permission_headers(Req, ?MODULE), |
| 38 | + #context{}}. |
| 39 | + |
| 40 | +variances(Req, Context) -> |
| 41 | + {[<<"accept-encoding">>, <<"origin">>], Req, Context}. |
| 42 | + |
| 43 | +content_types_accepted(ReqData, Context) -> |
| 44 | + {[{{<<"application">>, <<"json">>, '*'}, accept_content}], ReqData, Context}. |
| 45 | + |
| 46 | +allowed_methods(ReqData, Context) -> |
| 47 | + {[<<"PUT">>, <<"OPTIONS">>], ReqData, Context}. |
| 48 | + |
| 49 | +resource_exists(ReqData, Context) -> |
| 50 | + %% just checking that the vhost requested exists |
| 51 | + {case rabbit_mgmt_util:all_or_one_vhost(ReqData, fun (_) -> [] end) of |
| 52 | + vhost_not_found -> false; |
| 53 | + _ -> true |
| 54 | + end, ReqData, Context}. |
| 55 | + |
| 56 | +is_authorized(ReqData, Context) -> |
| 57 | + rabbit_mgmt_util:is_authorized_vhost(ReqData, Context). |
| 58 | + |
| 59 | +accept_content(ReqData0, #context{user = #user{username = ActingUser}} = Context) -> |
| 60 | + %% TODO validate arguments? |
| 61 | + VHost = rabbit_mgmt_util:id(vhost, ReqData0), |
| 62 | + Name = rabbit_mgmt_util:id(name, ReqData0), |
| 63 | + rabbit_mgmt_util:with_decode( |
| 64 | + [], ReqData0, Context, |
| 65 | + fun([], BodyMap, ReqData) -> |
| 66 | + PartitionsBin = maps:get(partitions, BodyMap, undefined), |
| 67 | + BindingKeysStr = maps:get('binding-keys', BodyMap, undefined), |
| 68 | + case validate_partitions_or_binding_keys(PartitionsBin, BindingKeysStr, ReqData, Context) of |
| 69 | + ok -> |
| 70 | + Arguments = maps:get(arguments, BodyMap, #{}), |
| 71 | + Node = get_node(BodyMap), |
| 72 | + case PartitionsBin of |
| 73 | + undefined -> |
| 74 | + BindingKeys = binding_keys(BindingKeysStr), |
| 75 | + Streams = streams_from_binding_keys(Name, BindingKeys), |
| 76 | + create_super_stream(Node, VHost, Name, Streams, |
| 77 | + Arguments, BindingKeys, ActingUser, |
| 78 | + ReqData, Context); |
| 79 | + _ -> |
| 80 | + case validate_partitions(PartitionsBin, ReqData, Context) of |
| 81 | + Partitions when is_integer(Partitions) -> |
| 82 | + Streams = streams_from_partitions(Name, Partitions), |
| 83 | + RoutingKeys = routing_keys(Partitions), |
| 84 | + create_super_stream(Node, VHost, Name, Streams, |
| 85 | + Arguments, RoutingKeys, ActingUser, |
| 86 | + ReqData, Context); |
| 87 | + Error -> |
| 88 | + Error |
| 89 | + end |
| 90 | + end; |
| 91 | + Error -> |
| 92 | + Error |
| 93 | + end |
| 94 | + end). |
| 95 | + |
| 96 | +%%------------------------------------------------------------------- |
| 97 | +get_node(Props) -> |
| 98 | + case maps:get(<<"node">>, Props, undefined) of |
| 99 | + undefined -> node(); |
| 100 | + N -> rabbit_nodes:make( |
| 101 | + binary_to_list(N)) |
| 102 | + end. |
| 103 | + |
| 104 | +binding_keys(BindingKeysStr) -> |
| 105 | + [rabbit_data_coercion:to_binary( |
| 106 | + string:strip(K)) |
| 107 | + || K |
| 108 | + <- string:tokens( |
| 109 | + rabbit_data_coercion:to_list(BindingKeysStr), ",")]. |
| 110 | + |
| 111 | +routing_keys(Partitions) -> |
| 112 | + [integer_to_binary(K) || K <- lists:seq(0, Partitions - 1)]. |
| 113 | + |
| 114 | +streams_from_binding_keys(Name, BindingKeys) -> |
| 115 | + [list_to_binary(binary_to_list(Name) |
| 116 | + ++ "-" |
| 117 | + ++ binary_to_list(K)) |
| 118 | + || K <- BindingKeys]. |
| 119 | + |
| 120 | +streams_from_partitions(Name, Partitions) -> |
| 121 | + [list_to_binary(binary_to_list(Name) |
| 122 | + ++ "-" |
| 123 | + ++ integer_to_list(K)) |
| 124 | + || K <- lists:seq(0, Partitions - 1)]. |
| 125 | + |
| 126 | +create_super_stream(NodeName, VHost, SuperStream, Streams, Arguments, |
| 127 | + RoutingKeys, ActingUser, ReqData, Context) -> |
| 128 | + case rabbit_misc:rpc_call(NodeName, |
| 129 | + rabbit_stream_manager, |
| 130 | + create_super_stream, |
| 131 | + [VHost, |
| 132 | + SuperStream, |
| 133 | + Streams, |
| 134 | + Arguments, |
| 135 | + RoutingKeys, |
| 136 | + ActingUser], |
| 137 | + ?DEFAULT_RPC_TIMEOUT) of |
| 138 | + ok -> |
| 139 | + {true, ReqData, Context}; |
| 140 | + {error, Reason} -> |
| 141 | + rabbit_mgmt_util:bad_request(io_lib:format("~p", [Reason]), |
| 142 | + ReqData, Context) |
| 143 | + end. |
| 144 | + |
| 145 | +validate_partitions_or_binding_keys(undefined, undefined, ReqData, Context) -> |
| 146 | + rabbit_mgmt_util:bad_request("Must specify partitions or binding keys", ReqData, Context); |
| 147 | +validate_partitions_or_binding_keys(_, undefined, _, _) -> |
| 148 | + ok; |
| 149 | +validate_partitions_or_binding_keys(undefined, _, _, _) -> |
| 150 | + ok; |
| 151 | +validate_partitions_or_binding_keys(_, _, ReqData, Context) -> |
| 152 | + rabbit_mgmt_util:bad_request("Specify partitions or binding keys, not both", ReqData, Context). |
| 153 | + |
| 154 | +validate_partitions(PartitionsBin, ReqData, Context) -> |
| 155 | + try |
| 156 | + case rabbit_data_coercion:to_integer(PartitionsBin) of |
| 157 | + Int when Int < 1 -> |
| 158 | + rabbit_mgmt_util:bad_request("The partition number must be greater than 0", ReqData, Context); |
| 159 | + Int -> |
| 160 | + Int |
| 161 | + end |
| 162 | + catch |
| 163 | + _:_ -> |
| 164 | + rabbit_mgmt_util:bad_request("The partitions must be a number", ReqData, Context) |
| 165 | + end. |
0 commit comments