Some testing and fixing receiving messages over SSE
This commit is contained in:
+2
-2
@@ -55,8 +55,8 @@ Processes the envelope and updates UI."
|
||||
(with-current-buffer buf
|
||||
(noise-chat-list--render))))))
|
||||
|
||||
;; Override the default handler in noise-sse
|
||||
(setq noise-sse--handle-event #'noise-receive--handle-sse-event)
|
||||
;; Register the receive handler with the SSE client.
|
||||
(setq noise-sse--event-handler #'noise-receive--handle-sse-event)
|
||||
|
||||
;;;###autoload
|
||||
(defun noise-receive-start-sse ()
|
||||
|
||||
+103
-19
@@ -39,38 +39,118 @@
|
||||
"Parse an SSE EVENT-STRING into a hash table.
|
||||
Returns nil for empty or invalid events."
|
||||
(when (and event-string (not (string-empty-p event-string)))
|
||||
(let ((lines (split-string event-string "\n"))
|
||||
(let ((lines (split-string event-string "\r?\n"))
|
||||
(data-lines '()))
|
||||
(dolist (line lines)
|
||||
(when (string-prefix-p "data: " line)
|
||||
(push (substring line 6) data-lines)))
|
||||
(when (string-match "\\`data:\\(?: \\)?\\(.*\\)\\'" line)
|
||||
(push (match-string 1 line) data-lines)))
|
||||
(when data-lines
|
||||
(let ((json-str (mapconcat #'identity (nreverse data-lines) "")))
|
||||
(let ((json-str (mapconcat #'identity (nreverse data-lines) "\n")))
|
||||
(condition-case nil
|
||||
(json-parse-string json-str)
|
||||
(error nil)))))))
|
||||
|
||||
(defvar noise-sse--buffer ""
|
||||
"Buffer for partial SSE event data.")
|
||||
"Decoded SSE event data awaiting a complete event delimiter.")
|
||||
|
||||
(defvar noise-sse--http-buffer ""
|
||||
"Raw HTTP response data buffered until headers are consumed.")
|
||||
|
||||
(defvar noise-sse--chunk-buffer ""
|
||||
"Raw chunked HTTP body data buffered until a whole chunk arrives.")
|
||||
|
||||
(defvar noise-sse--headers-received nil
|
||||
"Non-nil once the HTTP response headers have been consumed.")
|
||||
|
||||
(defvar noise-sse--chunked-response-p nil
|
||||
"Non-nil when the SSE response uses chunked transfer encoding.")
|
||||
|
||||
(defun noise-sse--dispatch-events (string)
|
||||
"Append STRING to the SSE event buffer and dispatch complete events."
|
||||
(setq noise-sse--buffer (concat noise-sse--buffer string))
|
||||
(while (string-match "\r?\n\r?\n" noise-sse--buffer)
|
||||
(let ((event-end (match-beginning 0))
|
||||
(event-next (match-end 0)))
|
||||
(let ((event-str (substring noise-sse--buffer 0 event-end)))
|
||||
(setq noise-sse--buffer (substring noise-sse--buffer event-next))
|
||||
(when-let ((event (noise-sse--parse-event event-str)))
|
||||
(funcall noise-sse--event-handler event))))))
|
||||
|
||||
(defun noise-sse--decode-chunked (string)
|
||||
"Decode HTTP chunked transfer body STRING.
|
||||
Returns newly decoded SSE payload, buffering incomplete chunks."
|
||||
(setq noise-sse--chunk-buffer (concat noise-sse--chunk-buffer string))
|
||||
(let ((decoded '()))
|
||||
(catch 'stop
|
||||
(while t
|
||||
(unless (string-match "\r?\n" noise-sse--chunk-buffer)
|
||||
(throw 'stop nil))
|
||||
(let* ((line-end (match-beginning 0))
|
||||
(line-next (match-end 0))
|
||||
(size-line (car (split-string
|
||||
(substring noise-sse--chunk-buffer 0 line-end)
|
||||
";" t "[ \t]*"))))
|
||||
(unless (and size-line
|
||||
(string-match-p "\\`[0-9A-Fa-f]+\\'" size-line))
|
||||
(throw 'stop nil))
|
||||
(let* ((size (string-to-number size-line 16))
|
||||
(data-start line-next)
|
||||
(data-end (+ data-start size)))
|
||||
(when (= size 0)
|
||||
(setq noise-sse--chunk-buffer "")
|
||||
(throw 'stop nil))
|
||||
(when (> data-end (length noise-sse--chunk-buffer))
|
||||
(throw 'stop nil))
|
||||
(let ((tail-end
|
||||
(cond
|
||||
((and (<= (+ data-end 2) (length noise-sse--chunk-buffer))
|
||||
(string= "\r\n"
|
||||
(substring noise-sse--chunk-buffer data-end (+ data-end 2))))
|
||||
(+ data-end 2))
|
||||
((and (< data-end (length noise-sse--chunk-buffer))
|
||||
(eq (aref noise-sse--chunk-buffer data-end) ?\n))
|
||||
(1+ data-end))
|
||||
(t
|
||||
(throw 'stop nil)))))
|
||||
(push (substring noise-sse--chunk-buffer data-start data-end) decoded)
|
||||
(setq noise-sse--chunk-buffer
|
||||
(substring noise-sse--chunk-buffer tail-end)))))))
|
||||
(apply #'concat (nreverse decoded))))
|
||||
|
||||
(defun noise-sse--consume-http-body (string)
|
||||
"Consume raw HTTP response STRING and return decoded SSE payload, if any."
|
||||
(let ((body string))
|
||||
(unless noise-sse--headers-received
|
||||
(setq noise-sse--http-buffer (concat noise-sse--http-buffer string))
|
||||
(if (not (string-match "\r?\n\r?\n" noise-sse--http-buffer))
|
||||
(setq body nil)
|
||||
(let ((headers (substring noise-sse--http-buffer 0 (match-beginning 0)))
|
||||
(body-start (match-end 0)))
|
||||
(let ((case-fold-search t))
|
||||
(setq noise-sse--chunked-response-p
|
||||
(string-match-p "\ntransfer-encoding:[^\r\n]*chunked" headers)))
|
||||
(setq body (substring noise-sse--http-buffer body-start)
|
||||
noise-sse--http-buffer ""
|
||||
noise-sse--headers-received t))))
|
||||
(when body
|
||||
(if noise-sse--chunked-response-p
|
||||
(noise-sse--decode-chunked body)
|
||||
body))))
|
||||
|
||||
(defun noise-sse--filter (proc string)
|
||||
"Process filter for SSE connection.
|
||||
Buffers incoming STRING and processes complete events."
|
||||
(setq noise-sse--buffer (concat noise-sse--buffer string))
|
||||
(while (string-match "\n\n" noise-sse--buffer)
|
||||
(let ((event-end (match-beginning 0))
|
||||
(event-start 0))
|
||||
(let ((event-str (substring noise-sse--buffer event-start event-end)))
|
||||
(setq noise-sse--buffer
|
||||
(substring noise-sse--buffer (+ event-end 2)))
|
||||
(when-let ((event (noise-sse--parse-event event-str)))
|
||||
(noise-sse--handle-event event))))))
|
||||
Consumes HTTP framing from STRING and dispatches decoded SSE events."
|
||||
(ignore proc)
|
||||
(when-let ((body (noise-sse--consume-http-body string)))
|
||||
(noise-sse--dispatch-events body)))
|
||||
|
||||
(defun noise-sse--handle-event (event)
|
||||
"Handle a parsed SSE EVENT.
|
||||
This function will be overridden in noise-receive to process envelopes."
|
||||
(defun noise-sse--default-handle-event (event)
|
||||
"Handle a parsed SSE EVENT when no receiver is installed yet."
|
||||
(message "SSE event received: %s" event))
|
||||
|
||||
(defvar noise-sse--event-handler #'noise-sse--default-handle-event
|
||||
"Function called with each parsed SSE event.")
|
||||
|
||||
(defvar noise-signal-cli-address)
|
||||
(defvar noise-account)
|
||||
|
||||
@@ -133,7 +213,11 @@ Handles connection close and triggers reconnection."
|
||||
(when noise-sse--reconnect-timer
|
||||
(cancel-timer noise-sse--reconnect-timer)
|
||||
(setq noise-sse--reconnect-timer nil))
|
||||
(setq noise-sse--buffer "")
|
||||
(setq noise-sse--buffer ""
|
||||
noise-sse--http-buffer ""
|
||||
noise-sse--chunk-buffer ""
|
||||
noise-sse--headers-received nil
|
||||
noise-sse--chunked-response-p nil)
|
||||
(message "Noise: SSE disconnected"))
|
||||
|
||||
(provide 'noise-sse)
|
||||
|
||||
@@ -189,3 +189,22 @@
|
||||
"timestamp" 1717930000000
|
||||
"message" "new message here")))
|
||||
(should (string-match-p "Sarah> new message here" (buffer-string))))))
|
||||
|
||||
(ert-deftest noise-receive-handle-sse-event-stores-message ()
|
||||
"An SSE envelope event stores the message and refreshes the open buffer."
|
||||
(noise-receive-test--with-db
|
||||
(noise-db-upsert-conversation "+1" :type "direct" :name "Sarah"
|
||||
:members '("+1"))
|
||||
(require 'noise-conversation)
|
||||
(with-current-buffer (noise-conversation-open "+1")
|
||||
(noise-receive--handle-sse-event
|
||||
(noise-receive-test--ht
|
||||
"envelope"
|
||||
(noise-receive-test--ht
|
||||
"sourceNumber" "+1" "sourceName" "Sarah"
|
||||
"dataMessage" (noise-receive-test--ht
|
||||
"timestamp" 1717930000000
|
||||
"message" "from sse"))))
|
||||
(should (string-match-p "Sarah> from sse" (buffer-string))))
|
||||
(let ((msg (car (noise-db-get-messages "+1"))))
|
||||
(should (string= "from sse" (gethash "body" msg))))))
|
||||
|
||||
+64
-15
@@ -60,24 +60,73 @@
|
||||
|
||||
(ert-deftest noise-sse--filter-processes-complete-event ()
|
||||
"Process filter parses and calls handler for complete events."
|
||||
(let ((processed-events '()))
|
||||
(cl-letf (((symbol-function 'noise-sse--handle-event)
|
||||
(lambda (event) (push event processed-events))))
|
||||
(let ((noise-sse--buffer ""))
|
||||
(noise-sse--filter nil "data: {\"type\":\"message\"}\n\n")
|
||||
(should (= 1 (length processed-events)))
|
||||
(should (hash-table-p (car processed-events)))))))
|
||||
(let ((processed-events '())
|
||||
(noise-sse--buffer "")
|
||||
(noise-sse--http-buffer "")
|
||||
(noise-sse--chunk-buffer "")
|
||||
(noise-sse--headers-received t)
|
||||
(noise-sse--chunked-response-p nil))
|
||||
(let ((noise-sse--event-handler
|
||||
(lambda (event) (push event processed-events))))
|
||||
(noise-sse--filter nil "data: {\"type\":\"message\"}\n\n")
|
||||
(should (= 1 (length processed-events)))
|
||||
(should (hash-table-p (car processed-events))))))
|
||||
|
||||
(ert-deftest noise-sse--filter-buffers-incomplete-event ()
|
||||
"Process filter buffers incomplete events until newline."
|
||||
(let ((processed-events '()))
|
||||
(cl-letf (((symbol-function 'noise-sse--handle-event)
|
||||
(lambda (event) (push event processed-events))))
|
||||
(let ((noise-sse--buffer ""))
|
||||
(noise-sse--filter nil "data: {\"type\":\"message\"")
|
||||
(should (= 0 (length processed-events)))
|
||||
(noise-sse--filter nil "}\n\n")
|
||||
(should (= 1 (length processed-events)))))))
|
||||
(let ((processed-events '())
|
||||
(noise-sse--buffer "")
|
||||
(noise-sse--http-buffer "")
|
||||
(noise-sse--chunk-buffer "")
|
||||
(noise-sse--headers-received t)
|
||||
(noise-sse--chunked-response-p nil))
|
||||
(let ((noise-sse--event-handler
|
||||
(lambda (event) (push event processed-events))))
|
||||
(noise-sse--filter nil "data: {\"type\":\"message\"")
|
||||
(should (= 0 (length processed-events)))
|
||||
(noise-sse--filter nil "}\n\n")
|
||||
(should (= 1 (length processed-events))))))
|
||||
|
||||
(ert-deftest noise-sse--filter-skips-http-headers ()
|
||||
"HTTP response headers should not block the first SSE event."
|
||||
(let ((processed-events '())
|
||||
(noise-sse--buffer "")
|
||||
(noise-sse--http-buffer "")
|
||||
(noise-sse--chunk-buffer "")
|
||||
(noise-sse--headers-received nil)
|
||||
(noise-sse--chunked-response-p nil))
|
||||
(let ((noise-sse--event-handler
|
||||
(lambda (event) (push event processed-events))))
|
||||
(noise-sse--filter nil
|
||||
(concat "HTTP/1.1 200 OK\r\n"
|
||||
"Content-Type: text/event-stream\r\n\r\n"
|
||||
"data: {\"type\":\"message\"}\n\n"))
|
||||
(should (= 1 (length processed-events)))
|
||||
(should (equal "message" (gethash "type" (car processed-events)))))))
|
||||
|
||||
(ert-deftest noise-sse--filter-decodes-chunked-http-body ()
|
||||
"Chunked HTTP SSE responses should be decoded before parsing events."
|
||||
(let ((processed-events '())
|
||||
(noise-sse--buffer "")
|
||||
(noise-sse--http-buffer "")
|
||||
(noise-sse--chunk-buffer "")
|
||||
(noise-sse--headers-received nil)
|
||||
(noise-sse--chunked-response-p nil))
|
||||
(let ((noise-sse--event-handler
|
||||
(lambda (event) (push event processed-events))))
|
||||
(noise-sse--filter nil
|
||||
(concat "HTTP/1.1 200 OK\r\n"
|
||||
"Transfer-Encoding: chunked\r\n"
|
||||
"Content-Type: text/event-stream\r\n\r\n"
|
||||
"12\r\n"
|
||||
"data: {\"type\":\"msg"
|
||||
"\r\n"
|
||||
"4\r\n"
|
||||
"\"}\n\n"
|
||||
"\r\n"
|
||||
"0\r\n\r\n"))
|
||||
(should (= 1 (length processed-events)))
|
||||
(should (equal "msg" (gethash "type" (car processed-events)))))))
|
||||
|
||||
(ert-deftest noise-sse--build-url ()
|
||||
"Build SSE endpoint URL from base address."
|
||||
|
||||
Reference in New Issue
Block a user