diff --git a/apps/smf_aaa/src/smf_aaa_charging.erl b/apps/smf_aaa/src/smf_aaa_charging.erl index f8f90ede..728c5313 100644 --- a/apps/smf_aaa/src/smf_aaa_charging.erl +++ b/apps/smf_aaa/src/smf_aaa_charging.erl @@ -6,6 +6,7 @@ -export([new/2, gy_ccr_initial/4, gy_ccr_update/4, gy_ccr_terminate/4, + gy_ccr_initial_issue/4, rf_initial/4, rf_update/4, rf_terminate/4, terminate/3, handle_reply/4]). @@ -33,6 +34,13 @@ gy_ccr_update(Ctx, Session, SOpts, Opts) -> gy_ccr_terminate(Ctx, Session, SOpts, Opts) -> invoke(Ctx, Session, SOpts, {gy, 'CCR-Terminate'}, Opts). +%% gy_ccr_initial_issue/4 — the async_m counterpart of gy_ccr_initial/4: runs +%% the CCR-Initial pre-send half on the Gy handler and returns the Promise +%% instead of blocking for the CCA. Session_at_send/State_at_send ride the +%% async_m do-block across the await and get folded by smf_aaa_ro:fold_cca/5. +gy_ccr_initial_issue(Ctx, Session, SOpts, Opts) -> + issue(Ctx, Session, SOpts, {gy, 'CCR-Initial'}, Opts). + rf_initial(Ctx, Session, SOpts, Opts) -> invoke(Ctx, Session, SOpts, {rf, 'Initial'}, Opts). @@ -71,6 +79,20 @@ invoke(#charging_ctx{app_id = AppId, handlers = Handlers0} = Ctx, {Result, Ctx#charging_ctx{handlers = Handlers1}, Session2, Events} end. +%% issue/5 — single-handler counterpart of invoke/5 for the async_m +%% pre-send path: no pipeline fan-out, no Events accumulator — just the one +%% handler's CCR-Initial pre-send + send_request, returning its Promise. +issue(#charging_ctx{app_id = AppId, handlers = Handlers0}, Session0, SOpts, Procedure, Opts0) -> + Opts = Opts0#{now => maps:get(now, Opts0, erlang:monotonic_time())}, + Session1 = smf_aaa_session:session_merge(Session0, SOpts), + #{procedures := Procedures} = smf_aaa:get_application(AppId), + [SvcOpts = #{service := Service} | _] = smf_aaa_session:get_services(Procedure, Procedures), + Svc = smf_aaa:get_service(Service), + StepOpts = maps:merge(Opts, maps:merge(Svc, SvcOpts)), + Handler = maps:get(handler, Svc), + HState0 = maps:get(Handler, Handlers0, undefined), + Handler:ccr_initial_issue(Session1, StepOpts, HState0). + run_pipeline(_Procedure, [], Session, Events, _Opts, Handlers) -> {ok, Session, Events, Handlers}; run_pipeline(Procedure, [#{service := Service} = SvcOpts | Tail], Session0, Events0, Opts, Handlers0) -> diff --git a/apps/smf_aaa/src/smf_aaa_gx.erl b/apps/smf_aaa/src/smf_aaa_gx.erl index 3143f081..92fbd831 100644 --- a/apps/smf_aaa/src/smf_aaa_gx.erl +++ b/apps/smf_aaa/src/smf_aaa_gx.erl @@ -24,6 +24,9 @@ initialize_handler/1, initialize_service/2, invoke/6, handle_response/6]). -export([to_session/3, from_session/2]). +%% async diameter bridge (see smf_aaa_pcf.erl / async_m) +-export([ccr_initial_issue/3, ccr_update_issue/3, fold_cca/5]). + -export([get_state_atom/1]). -ignore_xref([from_session/2, get_state_atom/1]). @@ -105,18 +108,12 @@ invoke(_Service, init, Session, Events, _Opts, _State) -> invoke(_Service, {_, 'CCR-Initial'}, Session0, Events, Opts, #state{state = stopped} = State0) -> - State = inc_request_number(State0#state{state = started}), - Keys = ['InPackets', 'OutPackets', 'InOctets', 'OutOctets', 'Acct-Session-Time'], - Session = maps:without(Keys, Session0), - RecType = ?'DIAMETER_GX_CC-REQUEST-TYPE_INITIAL_REQUEST', - Request = make_CCR(RecType, Session, Opts, State), + {Request, Session, State} = ccr_initial_pre_send(Session0, Opts, State0), await_response(send_request(Request, Opts), Session, Events, State, Opts); -invoke(_Service, {_, 'CCR-Update'}, Session, Events, Opts, +invoke(_Service, {_, 'CCR-Update'}, Session0, Events, Opts, #state{state = started} = State0) -> - State = inc_request_number(State0), - RecType = ?'DIAMETER_GX_CC-REQUEST-TYPE_UPDATE_REQUEST', - Request = make_CCR(RecType, Session, Opts, State), + {Request, Session, State} = ccr_update_pre_send(Session0, Opts, State0), await_response(send_request(Request, Opts), Session, Events, State, Opts); invoke(_Service, {_, 'CCR-Terminate'}, Session, Events, Opts, @@ -165,6 +162,59 @@ handle_response(Promise, Msg, Session, Events, Opts, #state{pending = Promise} = handle_response(_Promise, _Msg, Session, Events, _Opts, State) -> {ok, Session, Events, State}. +%%%=================================================================== +%%% async_m diameter bridge +%%%=================================================================== + +%% pre-send half of CCR-Initial: #state{} transition + volume-key strip + +%% make_CCR. Shared by invoke/6 (blocking path) and ccr_initial_issue/3 +%% (async_m path) so the two can't diverge. +ccr_initial_pre_send(Session0, Opts, State0) -> + State = inc_request_number(State0#state{state = started}), + Keys = ['InPackets', 'OutPackets', 'InOctets', 'OutOctets', 'Acct-Session-Time'], + Session = maps:without(Keys, Session0), + RecType = ?'DIAMETER_GX_CC-REQUEST-TYPE_INITIAL_REQUEST', + Request = make_CCR(RecType, Session, Opts, State), + {Request, Session, State}. + +%% ccr_initial_issue/3 — the async_m counterpart of invoke/6's CCR-Initial +%% clause: runs the same pre-send steps, but hands back the Promise instead +%% of blocking on await_response/parking it in #state.pending. +-spec ccr_initial_issue(map(), map(), #state{}) -> {reference(), map(), #state{}}. +ccr_initial_issue(Session0, Opts, State0) -> + {Request, Session, State} = ccr_initial_pre_send(Session0, Opts, State0), + Promise = send_request(Request, Opts), + {Promise, Session, State}. + +%% pre-send half of CCR-Update: #state{} transition + make_CCR. Shared by +%% invoke/6 (blocking path) and ccr_update_issue/3 (async_m path) so the +%% two can't diverge. +ccr_update_pre_send(Session, Opts, State0) -> + State = inc_request_number(State0), + RecType = ?'DIAMETER_GX_CC-REQUEST-TYPE_UPDATE_REQUEST', + Request = make_CCR(RecType, Session, Opts, State), + {Request, Session, State}. + +%% ccr_update_issue/3 — the async_m counterpart of invoke/6's CCR-Update +%% clause: runs the same pre-send steps, but hands back the Promise instead +%% of blocking on await_response/parking it in #state.pending. +-spec ccr_update_issue(map(), map(), #state{}) -> {reference(), map(), #state{}}. +ccr_update_issue(Session0, Opts, State0) -> + {Request, Session, State} = ccr_update_pre_send(Session0, Opts, State0), + Promise = send_request(Request, Opts), + {Promise, Session, State}. + +%% fold_cca/5 — the async_m counterpart of await_response's blocking +%% handle_cca call: fold a raw CCA (or diameter error) into the session and +%% events, re-wrapping the resulting #state{} into a #pcf_ctx{} exactly as +%% the fire-and-forget handle_reply/4 path (smf_aaa_pcf.erl) does. +fold_cca(Msg, Session, Events, Opts, State) -> + {Result, Session1, Events1, State1} = handle_cca(Msg, Session, Events, Opts, State), + {Result, Session1, Events1, wrap_ctx(State1)}. + +wrap_ctx(State) -> + #pcf_ctx{handlers = #{?MODULE => State}}. + %%=================================================================== %% DIAMETER handler callbacks %%=================================================================== diff --git a/apps/smf_aaa/src/smf_aaa_pcf.erl b/apps/smf_aaa/src/smf_aaa_pcf.erl index e79b81db..e54bc5cd 100644 --- a/apps/smf_aaa/src/smf_aaa_pcf.erl +++ b/apps/smf_aaa/src/smf_aaa_pcf.erl @@ -6,6 +6,7 @@ -export([new/2, ccr_initial/4, ccr_update/4, ccr_terminate/4, + ccr_initial_issue/4, ccr_update_issue/4, terminate/3, handle_reply/4]). -ignore_xref([terminate/3]). @@ -32,6 +33,20 @@ ccr_update(Ctx, Session, SOpts, Opts) -> ccr_terminate(Ctx, Session, SOpts, Opts) -> invoke(Ctx, Session, SOpts, {gx, 'CCR-Terminate'}, Opts). +%% ccr_initial_issue/4 — the async_m counterpart of ccr_initial/4: runs the +%% CCR-Initial pre-send half on the Gx handler and returns the Promise +%% instead of blocking for the CCA. Session_at_send/State_at_send ride the +%% async_m do-block across the await and get folded by smf_aaa_gx:fold_cca/5. +ccr_initial_issue(Ctx, Session, SOpts, Opts) -> + issue(Ctx, Session, SOpts, {gx, 'CCR-Initial'}, Opts). + +%% ccr_update_issue/4 — the async_m counterpart of ccr_update/4: runs the +%% CCR-Update pre-send half on the Gx handler and returns the Promise +%% instead of blocking for the CCA. Session_at_send/State_at_send ride the +%% async_m do-block across the await and get folded by smf_aaa_gx:fold_cca/5. +ccr_update_issue(Ctx, Session, SOpts, Opts) -> + issue(Ctx, Session, SOpts, {gx, 'CCR-Update'}, Opts). + terminate(Ctx, Session, Opts) -> invoke(Ctx, Session, #{'Termination-Cause' => error}, terminate, Opts). @@ -60,6 +75,25 @@ invoke(#pcf_ctx{app_id = AppId, handlers = Handlers0} = Ctx, {Result, Ctx#pcf_ctx{handlers = Handlers1}, Session2, Events} end. +%% issue/5 — single-handler counterpart of invoke/5 for the async_m +%% pre-send path: no pipeline fan-out, no Events accumulator — just the one +%% handler's CCR-Initial/CCR-Update pre-send + send_request, returning its +%% Promise. +issue(#pcf_ctx{app_id = AppId, handlers = Handlers0}, Session0, SOpts, Procedure, Opts0) -> + Opts = Opts0#{now => maps:get(now, Opts0, erlang:monotonic_time())}, + Session1 = smf_aaa_session:session_merge(Session0, SOpts), + #{procedures := Procedures} = smf_aaa:get_application(AppId), + [SvcOpts = #{service := Service} | _] = smf_aaa_session:get_services(Procedure, Procedures), + Svc = smf_aaa:get_service(Service), + StepOpts = maps:merge(Opts, maps:merge(Svc, SvcOpts)), + Handler = maps:get(handler, Svc), + HState0 = maps:get(Handler, Handlers0, undefined), + IssueFun = issue_fun(Procedure), + Handler:IssueFun(Session1, StepOpts, HState0). + +issue_fun({_, 'CCR-Initial'}) -> ccr_initial_issue; +issue_fun({_, 'CCR-Update'}) -> ccr_update_issue. + run_pipeline(_Procedure, [], Session, Events, _Opts, Handlers) -> {ok, Session, Events, Handlers}; run_pipeline(Procedure, [#{service := Service} = SvcOpts | Tail], Session0, Events0, Opts, Handlers0) -> diff --git a/apps/smf_aaa/src/smf_aaa_ro.erl b/apps/smf_aaa/src/smf_aaa_ro.erl index 529c3031..4eba7871 100644 --- a/apps/smf_aaa/src/smf_aaa_ro.erl +++ b/apps/smf_aaa/src/smf_aaa_ro.erl @@ -23,6 +23,9 @@ -export([validate_handler/1, validate_service/3, validate_procedure/5, initialize_handler/1, initialize_service/2, invoke/6, handle_response/6]). -export([to_session/3, from_session/2]). + +%% async diameter bridge (see smf_aaa_charging.erl / async_m) +-export([ccr_initial_issue/3, fold_cca/5]). -export([get_state_atom/1]). -ignore_xref([from_session/2, get_state_atom/1]). @@ -110,11 +113,7 @@ invoke(_Service, init, Session, Events, _Opts, _State) -> invoke(_Service, {_, 'CCR-Initial'}, Session0, Events, Opts, #state{state = stopped} = State0) -> - State = inc_request_number(State0#state{state = started}), - Keys = ['InPackets', 'OutPackets', 'InOctets', 'OutOctets', 'Acct-Session-Time'], - Session = maps:without(Keys, Session0), - RecType = ?'DIAMETER_RO_CC-REQUEST-TYPE_INITIAL_REQUEST', - Request = make_CCR(RecType, Session, Opts, State), + {Request, Session, State} = ccr_initial_pre_send(Session0, Opts, State0), await_response(send_request(Request, Opts), Session, Events, State, Opts); invoke(_Service, {_, 'CCR-Update'}, Session, Events, Opts, @@ -191,6 +190,42 @@ handle_response(Promise, Msg, Session, Events, Opts, #state{pending = Promise} = handle_response(_Promise, _Msg, Session, Events, _Opts, State) -> {ok, Session, Events, State}. +%%%=================================================================== +%%% async_m diameter bridge +%%%=================================================================== + +%% pre-send half of CCR-Initial: #state{} transition + volume-key strip + +%% make_CCR (which itself builds the initial credit request). Shared by +%% invoke/6 (blocking path) and ccr_initial_issue/3 (async_m path) so the +%% two can't diverge. +ccr_initial_pre_send(Session0, Opts, State0) -> + State = inc_request_number(State0#state{state = started}), + Keys = ['InPackets', 'OutPackets', 'InOctets', 'OutOctets', 'Acct-Session-Time'], + Session = maps:without(Keys, Session0), + RecType = ?'DIAMETER_RO_CC-REQUEST-TYPE_INITIAL_REQUEST', + Request = make_CCR(RecType, Session, Opts, State), + {Request, Session, State}. + +%% ccr_initial_issue/3 — the async_m counterpart of invoke/6's CCR-Initial +%% clause: runs the same pre-send steps, but hands back the Promise instead +%% of blocking on await_response/parking it in #state.pending. +-spec ccr_initial_issue(map(), map(), #state{}) -> {reference(), map(), #state{}}. +ccr_initial_issue(Session0, Opts, State0) -> + {Request, Session, State} = ccr_initial_pre_send(Session0, Opts, State0), + Promise = send_request(Request, Opts), + {Promise, Session, State}. + +%% fold_cca/5 — the async_m counterpart of await_response's blocking +%% handle_cca call: fold a raw CCA (or diameter error) into the session and +%% events, re-wrapping the resulting #state{} into a #charging_ctx{} exactly +%% as the fire-and-forget handle_reply/4 path (smf_aaa_charging.erl) does. +fold_cca(Msg, Session, Events, Opts, State) -> + {Result, Session1, Events1, State1} = handle_cca(Msg, Session, Events, Opts, State), + {Result, Session1, Events1, wrap_ctx(State1)}. + +wrap_ctx(State) -> + #charging_ctx{handlers = #{?MODULE => State}}. + %%=================================================================== %% DIAMETER handler callbacks %%=================================================================== diff --git a/apps/smf_aaa/src/smf_aaa_static.erl b/apps/smf_aaa/src/smf_aaa_static.erl index 02550f6c..bd726278 100644 --- a/apps/smf_aaa/src/smf_aaa_static.erl +++ b/apps/smf_aaa/src/smf_aaa_static.erl @@ -13,6 +13,9 @@ -export([validate_handler/1, validate_service/3, validate_procedure/5, initialize_handler/1, initialize_service/2, invoke/6, handle_response/6]). +%% async_m mock API +-export([ccr_initial_issue/3, ccr_update_issue/3]). + -export([get_state_atom/1]). -ignore_xref([get_state_atom/1]). @@ -58,6 +61,44 @@ invoke(_Service, _Procedure, Session, Events, Opts, State) -> handle_response(_Promise, _Msg, Session, Events, _Opts, State) -> {ok, Session, Events, State}. +%% ccr_initial_issue/3 — async_m mock counterpart of invoke/6: resolves the +%% canned answer exactly as invoke/6 does, but instead of processing it +%% inline, hands back a Promise and delivers the raw CCA to Owner +%% (self() at call time), the same way a real Gx handler's issue/3 would. +-spec ccr_initial_issue(map(), map(), term()) -> {reference(), map(), term()}. +ccr_initial_issue(Session, Opts, State) -> + issue(Session, Opts, State). + +%% ccr_update_issue/3 — see ccr_initial_issue/3. +-spec ccr_update_issue(map(), map(), term()) -> {reference(), map(), term()}. +ccr_update_issue(Session, Opts, State) -> + issue(Session, Opts, State). + +%% issue/3 — shared by ccr_initial_issue/3 and ccr_update_issue/3: resolve +%% the canned answer's AVP map exactly as invoke/6 does (#{answers := +%% Answers, answer := Answer}), shape it as a raw ['CCA' | Avps] matching +%% smf_aaa_gx:handle_cca/5's pattern, and deliver it to Owner. +issue(Session, #{answers := Answers, answer := Answer} = Opts, State) -> + AVPs = + case Answers of + #{Answer := #{avps := A}} -> A; + _ -> #{} + end, + Promise = make_ref(), + Owner = self(), + Owner ! {'$reply', Promise, smf_aaa_gx, ['CCA' | to_raw_cca_avps(AVPs)], Opts}, + {Promise, Session, State}. + +%% to_raw_cca_avps/1 — shape a canned answer's AVP map as the raw CCA AVPs +%% smf_aaa_gx:handle_cca/5 pattern-matches: 'Result-Code' list-wrapped +%% ([RC]); everything else (including repeated AVPs like +%% Charging-Rule-Install/-Remove) passes through unchanged, since the +%% static config already list-wraps those. +to_raw_cca_avps(#{'Result-Code' := RC} = AVPs) when is_integer(RC) -> + AVPs#{'Result-Code' => [RC]}; +to_raw_cca_avps(AVPs) -> + AVPs. + %%%=================================================================== %%% Options Validation %%%=================================================================== diff --git a/apps/smf_aaa/test/smf_aaa_fold_SUITE.erl b/apps/smf_aaa/test/smf_aaa_fold_SUITE.erl new file mode 100644 index 00000000..a378825a --- /dev/null +++ b/apps/smf_aaa/test/smf_aaa_fold_SUITE.erl @@ -0,0 +1,57 @@ +%% Copyright 2026, Next-NF + +%% Tests for smf_aaa_gx:fold_cca/5 and smf_aaa_ro:fold_cca/5 — the async_m +%% counterpart of the blocking handle_cca fold done by await_response/5. + +-module(smf_aaa_fold_SUITE). + +-compile([export_all, nowarn_export_all]). + +-include_lib("common_test/include/ct.hrl"). +-include("../include/smf_aaa_session.hrl"). + +all() -> + [gx_install, gx_timeout, gy_update_credits, gy_timeout]. + +%%%=================================================================== +%%% Gx (smf_aaa_gx) +%%%=================================================================== + +gx_install(_Config) -> + {ok, Session0, [], State0} = smf_aaa_gx:invoke(gx, init, #{}, [], #{}, undefined), + CCA = ['CCA' | #{'Result-Code' => [2001], + 'Charging-Rule-Install' => + [#{'Charging-Rule-Base-Name' => [<<"m2m0001">>]}]}], + {ok, _Session1, Events, #pcf_ctx{}} = + smf_aaa_gx:fold_cca(CCA, Session0, [], #{}, State0), + {pcc, install, _} = lists:keyfind(pcc, 1, Events), + ok. + +gx_timeout(_Config) -> + {ok, Session0, [], State0} = smf_aaa_gx:invoke(gx, init, #{}, [], #{}, undefined), + {{error, timeout}, _Session1, _Events, #pcf_ctx{}} = + smf_aaa_gx:fold_cca({error, timeout}, Session0, [], #{}, State0), + ok. + +%%%=================================================================== +%%% Gy (smf_aaa_ro) +%%%=================================================================== + +gy_update_credits(_Config) -> + {ok, Session0a, [], State0} = smf_aaa_ro:invoke(gy, init, #{}, [], #{}, undefined), + Session0 = Session0a#{credits => #{1000 => empty}}, + CCA = ['CCA' | #{'Result-Code' => 2001, + 'Multiple-Services-Credit-Control' => + [#{'Rating-Group' => [1000], + 'Granted-Service-Unit' => [#{'CC-Time' => [3600]}], + 'Result-Code' => [2001]}]}], + {ok, _Session1, Events, #charging_ctx{}} = + smf_aaa_ro:fold_cca(CCA, Session0, [], #{}, State0), + {update_credits, _} = lists:keyfind(update_credits, 1, Events), + ok. + +gy_timeout(_Config) -> + {ok, Session0, [], State0} = smf_aaa_ro:invoke(gy, init, #{}, [], #{}, undefined), + {{error, timeout}, _Session1, _Events, #charging_ctx{}} = + smf_aaa_ro:fold_cca({error, timeout}, Session0, [], #{}, State0), + ok. diff --git a/apps/smf_aaa/test/smf_aaa_static_async_SUITE.erl b/apps/smf_aaa/test/smf_aaa_static_async_SUITE.erl new file mode 100644 index 00000000..32eac32c --- /dev/null +++ b/apps/smf_aaa/test/smf_aaa_static_async_SUITE.erl @@ -0,0 +1,43 @@ +-module(smf_aaa_static_async_SUITE). +-compile([export_all, nowarn_export_all]). +-include_lib("common_test/include/ct.hrl"). + +all() -> [ccr_update_async_loop]. + +ccr_update_async_loop(_Config) -> + %% a canned CCR-U answer that removes a rule (the Charging-Rule-Remove shape fold_cca turns into + %% a {pcc, remove, _} event). Build Opts/Session/State the way smf_aaa_pcf:issue/5 hands them to + %% the handler (read smf_aaa_static:invoke/6 + smf_aaa_pcf:issue/5 for the exact shapes). + {Session0, State0, Opts} = setup_ccru_remove(), + {Promise, _Session1, _State1} = smf_aaa_static:ccr_update_issue(Session0, Opts, State0), + RawCCA = receive + {'$reply', Promise, _Handler, Msg, _Opts} -> Msg + after 2000 -> ct:fail(no_reply) + end, + %% the delivered Msg is a raw ['CCA' | AVPs] that fold_cca consumes. + %% fold_cca/5 -> {Result, Session1, Events1, Ctx1} (see smf_aaa_gx.erl); + %% the brief's skeleton had Events/Ctx swapped, fixed here to match. + {ok, _Session2, Events, _Ctx} = smf_aaa_gx:fold_cca(RawCCA, Session0, [], Opts, State0), + true = lists:any(fun({pcc, remove, _}) -> true; (_) -> false end, Events), + ok. + +%% setup_ccru_remove/0 — build the {Session0, State0, Opts} triple the way +%% smf_aaa_pcf:issue/5 hands them to Handler:ccr_update_issue/3: +%% Session1 = the (merged) session map +%% StepOpts = Opts, including the canned #{answers, answer} that +%% smf_aaa_static:invoke/6 (and now issue/3) reads +%% HState0 = maps:get(Handler, Handlers0, undefined) — the mock has no +%% internal state of its own, so `undefined` is exactly what +%% a real pcf_ctx with no handler state yet would hand over. +setup_ccru_remove() -> + Session0 = #{}, + State0 = undefined, + Answers = + #{'Update-Gx-Remove' => + #{avps => + #{'Result-Code' => 2001, + 'Charging-Rule-Remove' => + [#{'Charging-Rule-Name' => [<<"m2m">>]}] + }}}, + Opts = #{answers => Answers, answer => 'Update-Gx-Remove'}, + {Session0, State0, Opts}. diff --git a/apps/smf_core/src/gtp_context.erl b/apps/smf_core/src/gtp_context.erl index f1a4cde0..30d788de 100644 --- a/apps/smf_core/src/gtp_context.erl +++ b/apps/smf_core/src/gtp_context.erl @@ -614,6 +614,15 @@ handle_event(info, #aaa_request{procedure = {_, 'RAR'}} = Request, _State, smf_aaa_session:aaa_reply(Request, {error, unknown_session}, #{}, Session), keep_state_and_data; +%% The diameter async bridge: smf_aaa_diameter_srv:send_request/4 posts the +%% same {'$reply', Promise, Handler, Msg, Opts} regardless of whether the +%% caller is the fire-and-forget AAA path (below) or an async_m procedure +%% parked on await(Promise) (Promise registered in async_pending). Route to +%% async_m first — it drains the specific ReqId; anything else falls through. +handle_event(info, {'$reply', Promise, _Handler, Msg, _Opts}, #{async_pending := P} = State, Data) + when is_map_key(Promise, P) -> + async_dispatch(Promise, Msg, State, Data); + handle_event(info, {'$reply', Promise, Handler, Msg, _Opts}, _State, Data) -> {Data1, Events} = handle_aaa_reply(Handler, Promise, Msg, Data), case Events of diff --git a/apps/smf_core/src/smf_pfcp_context.erl b/apps/smf_core/src/smf_pfcp_context.erl index 1ad3ea1a..16dc6073 100644 --- a/apps/smf_core/src/smf_pfcp_context.erl +++ b/apps/smf_core/src/smf_pfcp_context.erl @@ -11,6 +11,8 @@ -compile({parse_transform, cut}). -export([create_session/5, + create_session_async/5, + create_session_result/4, modify_session/5, modify_session_async/5, modify_session_result/2, @@ -209,6 +211,45 @@ session_establishment_request(Handler, PCC, PCtx0, BearerMap0, Ctx) -> {error, ?CTX_ERR(?FATAL, system_failure)} end. +%% create_session_async/5 — same rule build as create_session/5, but issues the request +%% asynchronously and returns a request id instead of blocking. +create_session_async(Handler, PCC, PCtx0, BearerMap0, Ctx) + when is_record(PCC, pcc_ctx) -> + register_ctx_ids(Handler, BearerMap0, PCtx0), + {ok, CntlNode, _, _} = smf_sx_socket:id(), + + PCtx1 = pctx_update_from_ctx(PCtx0, Ctx), + {SxRules, SxErrors, PCtx2} = build_sx_rules(PCC, #{}, PCtx1, BearerMap0), + ?LOG(debug, "SxRules: ~p~n", [SxRules]), + ?LOG(debug, "SxErrors: ~p~n", [SxErrors]), + ?LOG(debug, "CtxPending: ~p~n", [Ctx]), + + IEs0 = pfcp_pctx_update(PCtx2, PCtx0, SxRules), + IEs1 = update_m_rec(smf_pfcp:f_seid(PCtx2, CntlNode), IEs0), + IEs = pfcp_user_id(Ctx, IEs1), + ?LOG(debug, "IEs: ~p~n", [IEs]), + + session_establishment_request_async(PCtx2, IEs). + +session_establishment_request_async(PCtx, IEs) -> + Req = #pfcp{version = v1, type = session_establishment_request, seq_no = 0, ie = IEs}, + ReqId = smf_sx_node:send_request(PCtx, Req), + {ok, {request, ReqId, PCtx}}. + +%% create_session_result/4 — decode the async reply into the same shape create_session/5 +%% returns. Unlike modify, accept must thread BearerMap and re-register ctx ids using the +%% response's F-TEIDs, mirroring what session_establishment_request/5 does on accept. +create_session_result(#pfcp{version = v1, type = session_establishment_response, + ie = #{pfcp_cause := 'Request accepted', + f_seid := #f_seid{}} = RespIEs}, + Handler, BearerMap0, PCtx0) -> + SessionInfo = session_info(RespIEs), + {BearerMap, PCtx} = update_bearer(RespIEs, BearerMap0, PCtx0), + register_ctx_ids(Handler, BearerMap, PCtx), + {ok, {update_dp_seid(RespIEs, PCtx), BearerMap, SessionInfo}}; +create_session_result(_Other, _Handler, _BearerMap0, _PCtx0) -> + {error, ?CTX_ERR(?FATAL, system_failure)}. + %% session_modification_request/2 session_modification_request(PCtx, ReqIEs) when ?is_non_empty_opts(ReqIEs) -> diff --git a/apps/smf_core/test/smf_pfcp_context_SUITE.erl b/apps/smf_core/test/smf_pfcp_context_SUITE.erl index 195b81f6..093d6814 100644 --- a/apps/smf_core/test/smf_pfcp_context_SUITE.erl +++ b/apps/smf_core/test/smf_pfcp_context_SUITE.erl @@ -4,7 +4,40 @@ -include_lib("pfcplib/include/pfcp_packet.hrl"). -include("../include/smf.hrl"). -all() -> [result_accept, result_wrong_cause, result_timeout, result_unreachable, result_dead]. +all() -> [result_accept, result_wrong_cause, result_timeout, result_unreachable, result_dead, + create_result_accept, create_result_wrong_cause, create_result_timeout]. + +create_result_accept(_Config) -> + meck:new(gtp_context_reg, [passthrough, no_link]), + meck:expect(gtp_context_reg, register, fun(_Keys, _Handler, _Pid) -> ok end), + try + PdrId = 1, + Key = {'Access', 1}, + PCtx = #pfcp_ctx{seid = #seid{cp = 1}, chid_by_pdr = #{PdrId => Key}}, + BearerMap = #{Key => #bearer{interface = 'Access', local = #fq_teid{ip = v4}}}, + Reply = #pfcp{version = v1, type = session_establishment_response, + ie = #{pfcp_cause => 'Request accepted', + f_seid => #f_seid{seid = 2, ipv4 = <<127,0,0,1>>}, + created_pdr => + [#{pdr_id => #pdr_id{id = PdrId}, + f_teid => #f_teid{teid = 16#1234, ipv4 = <<127,0,0,1>>}}]}}, + {ok, {#pfcp_ctx{}, _BearerMap, _SessionInfo}} = + smf_pfcp_context:create_session_result(Reply, test_handler, BearerMap, PCtx), + ok + after + meck:unload(gtp_context_reg) + end. + +create_result_wrong_cause(_Config) -> + PCtx = #pfcp_ctx{}, + Reply = #pfcp{version = v1, type = session_establishment_response, + ie = #{pfcp_cause => 'System failure'}}, + {error, _} = smf_pfcp_context:create_session_result(Reply, test_handler, #{}, PCtx), + ok. + +create_result_timeout(_Config) -> + {error, _} = smf_pfcp_context:create_session_result(timeout, test_handler, #{}, #pfcp_ctx{}), + ok. result_accept(_Config) -> PCtx = #pfcp_ctx{},