feat: add client mqueue/inflight messages API #12561

SergeTupchiy · 2024-02-21T17:59:52Z

Fixes EMQX-11861
Release version: v/e5.6.0

NOTE: session_mem only implementation, persistent_session is planned to be addressed in a separate PR.

Summary

PR Checklist

Please convert it to a draft if any of the following conditions are not met. Reviewers may skip over until all the items are checked:

Added tests for the changes
Added property-based tests for code which performs user input validation
Changed lines covered in coverage report
Change log has been added to changes/(ce|ee)/(feat|perf|fix|breaking)-<PR-id>.en.md files
For internal contributor: there is a jira ticket to track this change
Created PR to emqx-docs if documentation update is required, or link to a follow-up jira ticket
Schema changes are backward compatible

Checklist for CI (.github/workflows) changes

If changed package build workflow, pass this action (manual trigger)
Change log has been added to changes/ dir for user-facing artifacts update

SergeTupchiy · 2024-02-22T17:43:22Z

apps/emqx_management/src/emqx_mgmt_api_clients.erl

+        from_clientid => emqx_utils_conv:bin(From),
+        from_username => maps:get(username, Headers, <<>>)
+    },
+    case erlang:byte_size(Payload) =< ?MAX_MSG_PAYLOAD_SIZE of


NOTE: message formatter/schema, payload size limit is copied from emqx_retainer_api.
Perhaps some extra info should be added when payload is omitted due to exceeding the limit?

Maybe add an option like ?payload=none | base64 | plain to the API.
Return 400 if payload is not JSON serializable and plain is requested.

About size limit, maybe alongside the page limit, there can be a max_payload_bytes limit and each response try to accumulate not greater than this limit however ensure at least one message is returned even if the size is above limit.

SergeTupchiy · 2024-02-22T17:44:44Z

apps/emqx/src/emqx_session_mem.erl

@@ -268,6 +268,9 @@ info(inflight_cnt, #session{inflight = Inflight}) ->
    emqx_inflight:size(Inflight);
 info(inflight_max, #session{inflight = Inflight}) ->
    emqx_inflight:max_size(Inflight);
+info({inflight_msgs, PagerParams}, #session{inflight = Inflight}) ->
+    {InflightList, Meta} = emqx_inflight:query(Inflight, PagerParams),
+    {[I#inflight_data.message || {_, I} <- InflightList], Meta};


Should we enrich message with some inflight details (wait_ack | wait_comp, timestamp)?

maybe no need for now

zmstone · 2024-02-26T17:29:48Z

apps/emqx_management/src/emqx_mgmt_api_clients.erl

+        qos => Qos,
+        topic => Topic,
+        publish_at =>
+            emqx_utils_calendar:epoch_to_rfc3339(Timestamp),


maybe should not format it.

I followed retainer messages formatting...

Removed time formatting

zmstone · 2024-02-26T17:34:58Z

apps/emqx_management/src/emqx_mgmt_api_clients.erl

+        from_clientid => emqx_utils_conv:bin(From),
+        from_username => maps:get(username, Headers, <<>>)
+    },
+    case erlang:byte_size(Payload) =< ?MAX_MSG_PAYLOAD_SIZE of


Maybe add an option like ?payload=none | base64 | plain to the API.
Return 400 if payload is not JSON serializable and plain is requested.

About size limit, maybe alongside the page limit, there can be a max_payload_bytes limit and each response try to accumulate not greater than this limit however ensure at least one message is returned even if the size is above limit.

zmstone · 2024-03-06T12:27:13Z

apps/emqx_management/src/emqx_mgmt_api_clients.erl

+                        <<"meta">> => #{
+                            <<"count">> => 1,
+                            <<"continuation">> => <<"AAYS53qRa0n07AAABFIACg==">>,
+                            <<"limit">> => 50,


there is maybe no need to return limit in response ?

Removed limit from meta

apps/emqx_management/src/emqx_mgmt_api_clients.erl

zmstone · 2024-03-06T12:29:57Z

changes/ce/feat-12561.en.md

@@ -0,0 +1,3 @@
+Implement HTTP API to get the list of client's inflight and mqueue messages:


Suggested change

Implement HTTP API to get the list of client's inflight and mqueue messages:

Implement HTTP APIs to get the list of client's inflight and mqueue messages:

zmstone · 2024-03-06T12:30:26Z

changes/ce/feat-12561.en.md

@@ -0,0 +1,3 @@
+Implement HTTP API to get the list of client's inflight and mqueue messages:
+ - GET /clients/{clientid}/mqueue_messages?page=1&limit=100


page is now continuation ?

Fixed, thanks

zmstone · 2024-03-06T12:35:38Z

changes/ce/feat-12561.en.md

@@ -0,0 +1,3 @@
+Implement HTTP API to get the list of client's inflight and mqueue messages:
+ - GET /clients/{clientid}/mqueue_messages?page=1&limit=100
+ - GET /clients/{clientid}/inflight_messages?page=1&limit=100


Maybe add:

he ordering of the messages in each page?

how to start the first page

how to continue the next pages

what happens if continuation is invalid

SergeTupchiy · 2024-03-07T08:22:19Z

apps/emqx_management/src/emqx_mgmt_api_clients.erl

+    Resp = #{meta => Meta1, data => format_msgs(Msgs, PayloadFmt, MaxBytes)},
+    %% Make sure minirest won't set another content-type for self-encoded JSON response body
+    Headers = #{<<"content-type">> => <<"application/json">>},
+    case emqx_utils_json:safe_encode(Resp) of


Note: this can actually serialize pretty much every payload, as it is not parsed and validated.
For example, if message.payload is invalid JSON string (missing enclosing }), it won't be detected:

emqx_utils_json:safe_encode(#{message => <<"{\"foo\": \"bar\"">>}). {ok,<<"{\"message\":\"{\\\"foo\\\": \\\"bar\\\"\"}">>} ```

apps/emqx_management/src/emqx_mgmt_api_clients.erl

SergeTupchiy force-pushed the EMQX-11861-client-mqueue-inflight-API branch 2 times, most recently from 6a46eef to a870dc0 Compare February 22, 2024 17:37

SergeTupchiy commented Feb 22, 2024

View reviewed changes

SergeTupchiy force-pushed the EMQX-11861-client-mqueue-inflight-API branch 2 times, most recently from 1a76e0f to 60000af Compare February 23, 2024 14:43

SergeTupchiy marked this pull request as ready for review February 23, 2024 14:54

SergeTupchiy requested review from lafirest and a team as code owners February 23, 2024 14:54

zmstone reviewed Feb 26, 2024

View reviewed changes

SergeTupchiy force-pushed the EMQX-11861-client-mqueue-inflight-API branch from 60000af to e0d13ed Compare March 5, 2024 19:45

SergeTupchiy requested a review from JimMoen as a code owner March 5, 2024 19:45

zmstone reviewed Mar 6, 2024

View reviewed changes

apps/emqx_management/src/emqx_mgmt_api_clients.erl Show resolved Hide resolved

zmstone reviewed Mar 6, 2024

View reviewed changes

SergeTupchiy force-pushed the EMQX-11861-client-mqueue-inflight-API branch 2 times, most recently from a186ee3 to 5606257 Compare March 6, 2024 19:24

SergeTupchiy requested a review from kjellwinblad as a code owner March 6, 2024 19:24

SergeTupchiy changed the base branch from master to release-56 March 6, 2024 19:24

SergeTupchiy commented Mar 7, 2024

View reviewed changes

feat: add client mqueue/inflight messages API

a2e7616

SergeTupchiy force-pushed the EMQX-11861-client-mqueue-inflight-API branch from 5606257 to a2e7616 Compare March 7, 2024 10:06

zmstone reviewed Mar 7, 2024

View reviewed changes

apps/emqx_management/src/emqx_mgmt_api_clients.erl Show resolved Hide resolved

zmstone approved these changes Mar 7, 2024

View reviewed changes

SergeTupchiy merged commit 1f38813 into emqx:release-56 Mar 7, 2024
166 checks passed

SergeTupchiy mentioned this pull request Mar 19, 2024

EMQX-11861 client mqueue inflight api updates #12732

Merged

9 tasks

BrewTestBot mentioned this pull request Mar 28, 2024

emqx 5.6.0 Homebrew/homebrew-core#167392

Merged

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat: add client mqueue/inflight messages API #12561

feat: add client mqueue/inflight messages API #12561

SergeTupchiy commented Feb 21, 2024 •

edited

SergeTupchiy Feb 22, 2024 •

edited

zmstone Feb 26, 2024

SergeTupchiy Mar 7, 2024

SergeTupchiy Feb 22, 2024 •

edited

zmstone Mar 7, 2024

zmstone Feb 26, 2024

SergeTupchiy Feb 26, 2024

SergeTupchiy Mar 7, 2024

zmstone Feb 26, 2024

zmstone Mar 6, 2024

SergeTupchiy Mar 7, 2024

zmstone Mar 6, 2024

SergeTupchiy Mar 7, 2024

zmstone Mar 6, 2024

SergeTupchiy Mar 7, 2024

zmstone Mar 6, 2024

SergeTupchiy Mar 7, 2024

SergeTupchiy Mar 7, 2024

		@@ -0,0 +1,3 @@
		Implement HTTP API to get the list of client's inflight and mqueue messages:

	Implement HTTP API to get the list of client's inflight and mqueue messages:
	Implement HTTP APIs to get the list of client's inflight and mqueue messages:

		@@ -0,0 +1,3 @@
		Implement HTTP API to get the list of client's inflight and mqueue messages:
		- GET /clients/{clientid}/mqueue_messages?page=1&limit=100

feat: add client mqueue/inflight messages API #12561

feat: add client mqueue/inflight messages API #12561

Conversation

SergeTupchiy commented Feb 21, 2024 • edited

Summary

PR Checklist

Checklist for CI (.github/workflows) changes

SergeTupchiy Feb 22, 2024 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

SergeTupchiy Feb 22, 2024 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

SergeTupchiy commented Feb 21, 2024 •

edited

SergeTupchiy Feb 22, 2024 •

edited

SergeTupchiy Feb 22, 2024 •

edited