Merge "websocket: Consider pending SSL data when waiting for socket input"

include/asterisk/http_websocket.h

@@ -337,6 +337,20 @@ AST_OPTIONAL_API(void, ast_websocket_unref, (struct ast_websocket *session), {re
  */
 AST_OPTIONAL_API(int, ast_websocket_fd, (struct ast_websocket *session), { errno = ENOSYS; return -1;});
 
+/*!
+ * \brief Wait for the WebSocket session to be ready to be read.
+ * \since 16.8.0
+ * \since 17.2.0
+ *
+ * \param session Pointer to the WebSocket session
+ * \param timeout the number of milliseconds to wait
+ *
+ * \retval -1 if error occurred
+ * \retval 0 if the timeout expired
+ * \retval 1 if the WebSocket session is ready for reading
+ */
+AST_OPTIONAL_API(int, ast_websocket_wait_for_input, (struct ast_websocket *session, int timeout), { errno = ENOSYS; return -1; });
+
 /*!
  * \brief Get the remote address for a WebSocket connected session.
  *

include/asterisk/iostream.h

@@ -126,6 +126,20 @@ void ast_iostream_set_exclusive_input(struct ast_iostream *stream, int exclusive
  */
 int ast_iostream_get_fd(struct ast_iostream *stream);
 
+/*!
+ * \brief Wait for input on the iostream's file descriptor
+ * \since 16.8.0
+ * \since 17.2.0
+ *
+ * \param stream A pointer to an iostream
+ * \param timeout the number of milliseconds to wait
+ *
+ * \retval -1 if error occurred
+ * \retval 0 if the timeout expired
+ * \retval 1 if the stream is ready for reading
+ */
+int ast_iostream_wait_for_input(struct ast_iostream *stream, int timeout);
+
 /*!
  * \brief Make an iostream non-blocking.
  *

main/iostream.c

@@ -86,6 +86,20 @@ int ast_iostream_get_fd(struct ast_iostream *stream)
 	return stream->fd;
 }
 
+int ast_iostream_wait_for_input(struct ast_iostream *stream, int timeout)
+{
+#if defined(DO_SSL)
+	/* Because SSL is read in blocks, it's possible that the last time we read we
+	   got more than we asked for and it is now buffered inside OpenSSL. If that
+	   is the case, calling ast_wait_for_input() will block until the fd is ready
+	   for reading again, which might never happen. */
+	if (stream->ssl && SSL_pending(stream->ssl)) {
+		return 1;
+	}
+#endif
+	return ast_wait_for_input(stream->fd, timeout);
+}
+
 void ast_iostream_nonblock(struct ast_iostream *stream)
 {
 	ast_fd_set_flags(stream->fd, O_NONBLOCK);

res/res_http_websocket.c

@@ -427,6 +427,11 @@ int AST_OPTIONAL_API_NAME(ast_websocket_fd)(struct ast_websocket *session)
 	return session->closing ? -1 : ast_iostream_get_fd(session->stream);
 }
 
+int AST_OPTIONAL_API_NAME(ast_websocket_wait_for_input)(struct ast_websocket *session, int timeout)
+{
+	return session->closing ? -1 : ast_iostream_wait_for_input(session->stream, timeout);
+}
+
 struct ast_sockaddr * AST_OPTIONAL_API_NAME(ast_websocket_remote_address)(struct ast_websocket *session)
 {
 	return &session->remote_address;
@@ -545,8 +550,8 @@ static inline int ws_safe_read(struct ast_websocket *session, char *buf, size_t
 				break;
 			}
 		}
-		if (ast_wait_for_input(ast_iostream_get_fd(session->stream), 1000) < 0) {
-			ast_log(LOG_ERROR, "ast_wait_for_input returned err: %s\n", strerror(errno));
+		if (ast_iostream_wait_for_input(session->stream, 1000) < 0) {
+			ast_log(LOG_ERROR, "ast_iostream_wait_for_input returned err: %s\n", strerror(errno));
 			*opcode = AST_WEBSOCKET_OPCODE_CLOSE;
 			session->closing = 1;
 			ao2_unlock(session);
@@ -974,7 +979,7 @@ static void websocket_echo_callback(struct ast_websocket *session, struct ast_va
 		goto end;
 	}
 
-	while ((res = ast_wait_for_input(ast_websocket_fd(session), -1)) > 0) {
+	while ((res = ast_websocket_wait_for_input(session, -1)) > 0) {
 		char *payload;
 		uint64_t payload_len;
 		enum ast_websocket_opcode opcode;

res/res_pjsip_transport_websocket.c

@@ -392,7 +392,7 @@ static void websocket_cb(struct ast_websocket *session, struct ast_variable *par
 	transport = create_data.transport;
 	read_data.transport = transport;
 
-	while (ast_wait_for_input(ast_websocket_fd(session), -1) > 0) {
+	while (ast_websocket_wait_for_input(session, -1) > 0) {
 		enum ast_websocket_opcode opcode;
 		int fragmented;