ARI: Channels added to Stasis application during WebSocket creation ...

Prior to ASTERISK-24988, the WebSocket handshake was resolved before Stasis
applications were registered. This was done such that the WebSocket would be
ready when an application is registered. However, by creating the WebSocket
first, the client had the ability to make requests for the Stasis application
it thought had been created with the initial handshake request. The inevitable
conclusion of this scenario was the cart being put before the horse.

ASTERISK-24988 resolved half of the problem by ensuring that the applications
were created and registered with Stasis prior to completing the handshake
with the client. While this meant that Stasis was ready when the client
received the green-light from Asterisk, it also meant that the WebSocket was
not yet ready for Stasis to dispatch messages.

This patch introduces a message queuing mechanism for delaying messages from
Stasis applications while the WebSocket is being constructed. When the ARI
event processor receives the message from the WebSocket that it is being
created, the event processor instantiates an event session which contains a
message queue. It then tries to create and register the requested applications
with Stasis. Messages that are dispatched from Stasis between this point and
the point at which the event processor is notified the WebSocket is ready, are
stashed in the queue. Once the WebSocket has been built, the queue's messages
are dispatched in the order in which they were originally received and the
queue is concurrently cleared.

ASTERISK-25181 #close
Reported By: Matt Jordan

Change-Id: Iafef7b85a2e0bf78c114db4c87ffc3d16d671a17

2 files changed, 25 insertions, 4 deletions

diff --git a/rest-api-templates/ari_resource.h.mustache b/rest-api-templates/ari_resource.h.mustache
index d3f40b6bd..f28e832d1 100644
--- a/rest-api-templates/ari_resource.h.mustache
+++ b/rest-api-templates/ari_resource.h.mustache
@@ -97,14 +97,28 @@ void ast_ari_{{c_name}}_{{c_nickname}}(struct ast_variable *headers, struct ast_
  * {{{notes}}}
 {{/notes}}
  *
+ * \retval  0 success
+ * \retval -1 error
+ */
+int ast_ari_websocket_{{c_name}}_{{c_nickname}}_init(void);
+
+/*!
+ * \brief {{summary}}
+{{#notes}}
+ *
+ * {{{notes}}}
+{{/notes}}
+ *
  * \param ser HTTP TCP/TLS Server Session
  * \param headers HTTP headers
  * \param args Swagger parameters
+ * \param session_id The id of the current session.
  *
  * \retval 0 success
  * \retval non-zero error
  */
-int ast_ari_websocket_{{c_name}}_{{c_nickname}}_attempted(struct ast_tcptls_session_instance *ser, struct ast_variable *headers, struct ast_ari_{{c_name}}_{{c_nickname}}_args *args);
+int ast_ari_websocket_{{c_name}}_{{c_nickname}}_attempted(struct ast_tcptls_session_instance *ser,
+	struct ast_variable *headers, struct ast_ari_{{c_name}}_{{c_nickname}}_args *args, const char *session_id);
 
 /*!
  * \brief {{summary}}
@@ -116,8 +130,10 @@ int ast_ari_websocket_{{c_name}}_{{c_nickname}}_attempted(struct ast_tcptls_sess
  * \param session ARI WebSocket.
  * \param headers HTTP headers.
  * \param args Swagger parameters.
+ * \param session_id The id of the current session.
  */
-void ast_ari_websocket_{{c_name}}_{{c_nickname}}_established(struct ast_ari_websocket_session *session, struct ast_variable *headers, struct ast_ari_{{c_name}}_{{c_nickname}}_args *args);
+void ast_ari_websocket_{{c_name}}_{{c_nickname}}_established(struct ast_ari_websocket_session *session,
+	struct ast_variable *headers, struct ast_ari_{{c_name}}_{{c_nickname}}_args *args);
 {{/is_websocket}}
 {{/operations}}
 {{/apis}}
diff --git a/rest-api-templates/res_ari_resource.c.mustache b/rest-api-templates/res_ari_resource.c.mustache
index 7fe360e89..36ca035ee 100644
--- a/rest-api-templates/res_ari_resource.c.mustache
+++ b/rest-api-templates/res_ari_resource.c.mustache
@@ -137,7 +137,8 @@ fin: __attribute__((unused))
 }
 {{/is_req}}
 {{#is_websocket}}
-static int ast_ari_{{c_name}}_{{c_nickname}}_ws_attempted_cb(struct ast_tcptls_session_instance *ser, struct ast_variable *get_params, struct ast_variable *headers)
+static int ast_ari_{{c_name}}_{{c_nickname}}_ws_attempted_cb(struct ast_tcptls_session_instance *ser,
+	struct ast_variable *get_params, struct ast_variable *headers, const char *session_id)
 {
 	struct ast_ari_{{c_name}}_{{c_nickname}}_args args = {};
 {{#has_parameters}}
@@ -156,7 +157,7 @@ static int ast_ari_{{c_name}}_{{c_nickname}}_ws_attempted_cb(struct ast_tcptls_s
 
 {{> param_parsing}}
 
-	res = ast_ari_websocket_{{c_name}}_{{c_nickname}}_attempted(ser, headers, &args);
+	res = ast_ari_websocket_{{c_name}}_{{c_nickname}}_attempted(ser, headers, &args, session_id);
 
 fin: __attribute__((unused))
 	if (!response) {
@@ -255,6 +256,10 @@ static int load_module(void)
 {{#has_websocket}}
 	struct ast_websocket_protocol *protocol;
 
+	if (ast_ari_websocket_{{c_name}}_{{c_nickname}}_init() == -1) {
+		return AST_MODULE_LOAD_FAILURE;
+	}
+
 	{{full_name}}.ws_server = ast_websocket_server_create();
 	if (!{{full_name}}.ws_server) {
 		return AST_MODULE_LOAD_FAILURE;