diff --git a/VERSION b/VERSION index 6da28dd..8294c18 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -0.1.1 \ No newline at end of file +0.1.2 \ No newline at end of file diff --git a/include/libe3/e3_agent.hpp b/include/libe3/e3_agent.hpp index 4c71c4d..51dff05 100644 --- a/include/libe3/e3_agent.hpp +++ b/include/libe3/e3_agent.hpp @@ -277,13 +277,28 @@ class E3Agent { /** Send a SubscriptionDelete for a previously subscribed RAN function. */ ErrorCode unsubscribe(uint32_t ran_function_id); - /** Send a dApp control action to the RAN. */ + /** + * @brief Send a dApp control action to the RAN. + * + * @param out_message_id when non-null, receives the assigned E3-MessageID on + * success, so the caller can correlate a later ack/response back to + * this send. + */ ErrorCode send_control(uint32_t ran_function_id, uint32_t control_id, - std::vector action_data); + std::vector action_data, + uint32_t* out_message_id = nullptr); - /** Send a dApp report to the RAN. */ - ErrorCode send_report(uint32_t ran_function_id, std::vector report_data); + /** + * @brief Send a dApp report to the RAN. + * + * @param out_message_id when non-null, receives the assigned E3-MessageID on + * success, so the caller can correlate a later ack/response back to + * this send. + */ + ErrorCode send_report(uint32_t ran_function_id, + std::vector report_data, + uint32_t* out_message_id = nullptr); /** Send a ReleaseMessage. The dApp stays running until stop() is called. */ ErrorCode release(); diff --git a/include/libe3/e3_interface.hpp b/include/libe3/e3_interface.hpp index 8f7f17d..697411e 100644 --- a/include/libe3/e3_interface.hpp +++ b/include/libe3/e3_interface.hpp @@ -189,11 +189,16 @@ class E3Interface { std::optional periodicity = std::nullopt, uint32_t* out_request_id = nullptr); ErrorCode queue_subscription_delete(uint32_t ran_function_id); + // out_message_id, when non-null, receives the assigned E3-MessageID on + // success, so the caller can correlate a later ack/response back to this + // send. ErrorCode queue_dapp_control_action(uint32_t ran_function_id, uint32_t control_id, - std::vector action_data); + std::vector action_data, + uint32_t* out_message_id = nullptr); ErrorCode queue_dapp_report(uint32_t ran_function_id, - std::vector report_data); + std::vector report_data, + uint32_t* out_message_id = nullptr); ErrorCode queue_release_message(); // Block until the setup handshake completes (or times out / fails). diff --git a/include/libe3/types.hpp b/include/libe3/types.hpp index 197f86d..83d7c26 100644 --- a/include/libe3/types.hpp +++ b/include/libe3/types.hpp @@ -300,6 +300,9 @@ struct XAppControlAction { uint32_t dapp_identifier{0}; uint32_t ran_function_identifier{0}; std::vector xapp_control_data; + // Set by the inbound loop from the wrapping E3-PDU's id field; not + // serialized, so it carries no ASN.1/JSON/Protobuf grammar of its own. + uint32_t message_id{0}; }; /** diff --git a/src/core/e3_agent.cpp b/src/core/e3_agent.cpp index faf3d5c..a7ad271 100644 --- a/src/core/e3_agent.cpp +++ b/src/core/e3_agent.cpp @@ -426,17 +426,20 @@ ErrorCode E3Agent::unsubscribe(uint32_t ran_function_id) { ErrorCode E3Agent::send_control(uint32_t ran_function_id, uint32_t control_id, - std::vector action_data) { + std::vector action_data, + uint32_t* out_message_id) { if (impl_->config.role != E3Role::DAPP) return ErrorCode::STATE_ERROR; if (!impl_->interface || !impl_->interface->is_running()) return ErrorCode::NOT_INITIALIZED; return impl_->interface->queue_dapp_control_action( - ran_function_id, control_id, std::move(action_data)); + ran_function_id, control_id, std::move(action_data), out_message_id); } -ErrorCode E3Agent::send_report(uint32_t ran_function_id, std::vector report_data) { +ErrorCode E3Agent::send_report(uint32_t ran_function_id, + std::vector report_data, + uint32_t* out_message_id) { if (impl_->config.role != E3Role::DAPP) return ErrorCode::STATE_ERROR; if (!impl_->interface || !impl_->interface->is_running()) return ErrorCode::NOT_INITIALIZED; - return impl_->interface->queue_dapp_report(ran_function_id, std::move(report_data)); + return impl_->interface->queue_dapp_report(ran_function_id, std::move(report_data), out_message_id); } ErrorCode E3Agent::release() { diff --git a/src/core/e3_interface.cpp b/src/core/e3_interface.cpp index de8df4e..2c85494 100644 --- a/src/core/e3_interface.cpp +++ b/src/core/e3_interface.cpp @@ -1129,7 +1129,10 @@ void E3Interface::inbound_loop_dapp() { } case PduType::XAPP_CONTROL_ACTION: { auto* a = std::get_if(&pdu.choice); - if (a) handle_xapp_control_action(*a, seq); + if (a) { + a->message_id = pdu.message_id; + handle_xapp_control_action(*a, seq); + } break; } case PduType::MESSAGE_ACK: { @@ -1389,7 +1392,8 @@ ErrorCode E3Interface::queue_subscription_delete(uint32_t ran_function_id) { ErrorCode E3Interface::queue_dapp_control_action( uint32_t ran_function_id, uint32_t control_id, - std::vector action_data + std::vector action_data, + uint32_t* out_message_id ) { if (!dapp_state_) return ErrorCode::STATE_ERROR; auto id = dapp_id(); @@ -1402,18 +1406,24 @@ ErrorCode E3Interface::queue_dapp_control_action( a.control_identifier = control_id; a.action_data = std::move(action_data); pdu.choice = std::move(a); - pdu.message_id = generate_message_id(); + const uint32_t mid = generate_message_id(); + pdu.message_id = mid; // Mirrors E3Agent::send_indication: this is an emit path, so it stamps // EMIT_ENTER itself rather than leaving it to queue_outbound, which only // allocates enqueue_seq for a Pdu that arrives with none. pdu.enqueue_seq = latrec_seq_next(); latrec_tstamp(pdu.enqueue_seq, LATREC_EMIT_ENTER, latrec_ctx(), ran_function_id); - return queue_outbound(std::move(pdu)); + ErrorCode rc = queue_outbound(std::move(pdu)); + if (rc == ErrorCode::SUCCESS && out_message_id) { + *out_message_id = mid; + } + return rc; } ErrorCode E3Interface::queue_dapp_report( uint32_t ran_function_id, - std::vector report_data + std::vector report_data, + uint32_t* out_message_id ) { if (!dapp_state_) return ErrorCode::STATE_ERROR; auto id = dapp_id(); @@ -1425,8 +1435,13 @@ ErrorCode E3Interface::queue_dapp_report( r.ran_function_identifier = ran_function_id; r.report_data = std::move(report_data); pdu.choice = std::move(r); - pdu.message_id = generate_message_id(); - return queue_outbound(std::move(pdu)); + const uint32_t mid = generate_message_id(); + pdu.message_id = mid; + ErrorCode rc = queue_outbound(std::move(pdu)); + if (rc == ErrorCode::SUCCESS && out_message_id) { + *out_message_id = mid; + } + return rc; } ErrorCode E3Interface::queue_release_message() { diff --git a/swig/e3_dapp_session.cpp b/swig/e3_dapp_session.cpp index 7ca3fd1..b6f894d 100644 --- a/swig/e3_dapp_session.cpp +++ b/swig/e3_dapp_session.cpp @@ -83,6 +83,7 @@ DAppSession::DAppSession(libe3::E3Config config, std::size_t queue_capacity) { ev.dapp_id = a.dapp_identifier; ev.ran_function_id = a.ran_function_identifier; ev.payload = a.xapp_control_data; + ev.request_id = a.message_id; impl->enqueue(std::move(ev)); }); @@ -217,13 +218,26 @@ int DAppSession::unsubscribe(uint32_t ran_function_id) { int DAppSession::send_control(uint32_t ran_function_id, uint32_t control_id, std::vector action_data) { - return static_cast( - impl_->agent->send_control(ran_function_id, control_id, std::move(action_data))); + uint32_t message_id = 0; + ErrorCode rc = impl_->agent->send_control( + ran_function_id, control_id, std::move(action_data), &message_id); + // On success return the assigned message id (positive, 1..1000) so Python + // can correlate a later ack/response by id; on failure return the + // ErrorCode (all error codes are negative, SUCCESS is 0). + if (rc == ErrorCode::SUCCESS) { + return static_cast(message_id); + } + return static_cast(rc); } int DAppSession::send_report(uint32_t ran_function_id, std::vector report_data) { - return static_cast( - impl_->agent->send_report(ran_function_id, std::move(report_data))); + uint32_t message_id = 0; + ErrorCode rc = impl_->agent->send_report(ran_function_id, std::move(report_data), &message_id); + // Same success/failure encoding as send_control() above. + if (rc == ErrorCode::SUCCESS) { + return static_cast(message_id); + } + return static_cast(rc); } int DAppSession::send_message_ack(uint32_t request_id, int response_code) { diff --git a/swig/e3_dapp_session.hpp b/swig/e3_dapp_session.hpp index 5b008c8..3c76d3e 100644 --- a/swig/e3_dapp_session.hpp +++ b/swig/e3_dapp_session.hpp @@ -77,7 +77,7 @@ struct E3Event { uint32_t dapp_id{0}; ///< dApp identifier the message targets uint32_t ran_function_id{0}; ///< RAN function (indication / xApp control) uint32_t subscription_id{0}; ///< subscription id (subscription response) - uint32_t request_id{0}; ///< request id (subscription response / ack) + uint32_t request_id{0}; ///< request/message id (subscription response / ack / xApp control) int response_code{-1}; ///< 0=positive, 1=negative, -1=n/a std::vector payload; ///< opaque E3SM bytes (indication / xApp control) uint64_t trace_seq{0}; ///< set when queued; keys the [latrec] LQ-stage records @@ -156,10 +156,20 @@ class DAppSession { /** @brief Delete a previously created subscription. @return ErrorCode as int. */ int unsubscribe(uint32_t ran_function_id); - /** @brief Send a dApp control action to the RAN. @return ErrorCode as int. */ + /** + * @brief Send a dApp control action to the RAN. + * @return the assigned message id (positive, 1..1000) on success, so the + * caller can correlate a later ack/response by id; a negative + * ErrorCode on failure. + */ int send_control(uint32_t ran_function_id, uint32_t control_id, std::vector action_data); - /** @brief Send a dApp report to the RAN. @return ErrorCode as int. */ + /** + * @brief Send a dApp report to the RAN. + * @return the assigned message id (positive, 1..1000) on success, so the + * caller can correlate a later ack/response by id; a negative + * ErrorCode on failure. + */ int send_report(uint32_t ran_function_id, std::vector report_data); /** @brief Acknowledge a request. @param response_code 0=positive,1=negative. @return ErrorCode as int. */ int send_message_ack(uint32_t request_id, int response_code); diff --git a/tests/integration/test_role_pair_zmq_ipc.cpp b/tests/integration/test_role_pair_zmq_ipc.cpp index 190d31e..5df7c8a 100644 --- a/tests/integration/test_role_pair_zmq_ipc.cpp +++ b/tests/integration/test_role_pair_zmq_ipc.cpp @@ -105,6 +105,7 @@ class TestSimpleSM : public ServiceModel { ++control_count; last_control_id = action.control_identifier; last_action_size = action.action_data.size(); + last_request_message_id = request_message_id; Pdu ack = make_message_ack_pdu(request_message_id, ResponseCode::POSITIVE); return emit_outbound(std::move(ack)); } @@ -112,6 +113,7 @@ class TestSimpleSM : public ServiceModel { std::atomic control_count{0}; std::atomic last_control_id{0}; std::atomic last_action_size{0}; + std::atomic last_request_message_id{0}; private: std::atomic running_{false}; @@ -157,6 +159,7 @@ TEST(role_pair_full_handshake_indication_control_report_release) { std::condition_variable cv; int indications = 0; bool sub_resp_ok = false; + std::vector xapp_control_message_ids; E3Agent dapp(make_dapp_config(ep)); dapp.set_indication_handler([&](const IndicationMessage& msg) { @@ -172,6 +175,13 @@ TEST(role_pair_full_handshake_indication_control_report_release) { sub_resp_ok = (r.response_code == ResponseCode::POSITIVE); cv.notify_all(); }); + // #68: the E3-PDU's id, decoded into Pdu::message_id, must be forwarded + // into XAppControlAction::message_id by the inbound loop. + dapp.set_xapp_control_handler([&](const XAppControlAction& a) { + std::lock_guard lk(mu); + xapp_control_message_ids.push_back(a.message_id); + cv.notify_all(); + }); ASSERT_TRUE(dapp.start() == ErrorCode::SUCCESS); @@ -209,7 +219,11 @@ TEST(role_pair_full_handshake_indication_control_report_release) { // Send a control action and verify the SM saw it std::vector ctrl; ASSERT_TRUE(libe3_examples::encode_simple_control(42, ctrl)); - ASSERT_TRUE(dapp.send_control(1, 1, ctrl) == ErrorCode::SUCCESS); + // #67: send_control's out_message_id must be the same id the RAN sees as + // request_message_id in handle_control_action. + uint32_t sent_message_id = 0; + ASSERT_TRUE(dapp.send_control(1, 1, ctrl, &sent_message_id) == ErrorCode::SUCCESS); + ASSERT_TRUE(sent_message_id != 0); // Allow the SM up to 2s to receive and ack the control for (int i = 0; i < 40 && sm_ptr->control_count.load() == 0; ++i) { @@ -217,6 +231,21 @@ TEST(role_pair_full_handshake_indication_control_report_release) { } ASSERT_GE(sm_ptr->control_count.load(), 1); ASSERT_EQ(sm_ptr->last_control_id.load(), 1u); + ASSERT_EQ(sm_ptr->last_request_message_id.load(), sent_message_id); + + // #68: two RAN -> dApp xApp control relays must each carry a distinct, + // nonzero message id through to the dApp's registered handler. + auto dapp_id = dapp.dapp_id(); + ASSERT_TRUE(dapp_id.has_value()); + ASSERT_TRUE(ran.send_xapp_control(*dapp_id, TestSimpleSM::RAN_FUNCTION_ID, ctrl) == ErrorCode::SUCCESS); + ASSERT_TRUE(ran.send_xapp_control(*dapp_id, TestSimpleSM::RAN_FUNCTION_ID, ctrl) == ErrorCode::SUCCESS); + { + std::unique_lock lk(mu); + ASSERT_TRUE(cv.wait_for(lk, 3s, [&]() { return xapp_control_message_ids.size() >= 2; })); + } + ASSERT_TRUE(xapp_control_message_ids[0] != 0); + ASSERT_TRUE(xapp_control_message_ids[1] != 0); + ASSERT_TRUE(xapp_control_message_ids[0] != xapp_control_message_ids[1]); // Release + stop ASSERT_TRUE(dapp.release() == ErrorCode::SUCCESS);