diff --git a/noise-receive.el b/noise-receive.el index e14dce4..fb0d00d 100644 --- a/noise-receive.el +++ b/noise-receive.el @@ -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 () diff --git a/noise-sse.el b/noise-sse.el index 80d6151..3c0d711 100644 --- a/noise-sse.el +++ b/noise-sse.el @@ -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) diff --git a/test/noise-receive-test.el b/test/noise-receive-test.el index 9dec3dc..bf1bbcb 100644 --- a/test/noise-receive-test.el +++ b/test/noise-receive-test.el @@ -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)))))) diff --git a/test/noise-sse-test.el b/test/noise-sse-test.el index 71db8f4..773680f 100644 --- a/test/noise-sse-test.el +++ b/test/noise-sse-test.el @@ -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."