Allow for a map as an async http message [IMMUTANT-549, IMMUTANT-547]

We also now throw if you try to set the :status or :headers once the
send has already started or on a websocket.

project.clj

@@ -91,7 +91,7 @@
                         clj-http                   "1.0.1"
 
                         ;; org.projectodd.wunderboss  "0.4.1"
-                        org.projectodd.wunderboss  "1.x.incremental.224"
+                        org.projectodd.wunderboss  "1.x.incremental.225"
                         ;; org.projectodd.wunderboss  "0.5.1-SNAPSHOT"
 
                         org.immutant               :version

web/src/immutant/web/async.clj

@@ -19,7 +19,7 @@
   (:import [org.projectodd.wunderboss.web.async Channel Channel$OnComplete HttpChannel]
            [org.projectodd.wunderboss.web.async.websocket WebsocketChannel]
            [java.io File FileInputStream InputStream]
-           java.util.Arrays
+           [java.util Arrays Map]
            clojure.lang.ISeq))
 
 (defn ^:internal ^:no-doc streaming-body? [body]
@@ -72,7 +72,7 @@
 (defn ^:private finalize-channel-response
   [^Channel ch status headers]
   (when (and (instance? HttpChannel ch)
-          (not (.headersSent ^HttpChannel ch)))
+          (not (.sendStarted ^HttpChannel ch)))
     (let [orig-response (.get ch :response-map)]
       ((.get ch :set-status-fn) (or status (:status orig-response)))
       ((.get ch :set-headers-fn) (or headers (:headers orig-response))))))
@@ -115,6 +115,15 @@
   (dispatch-message [_ ch options]
     (wboss-send ch nil options))
 
+  Map
+  (dispatch-message [message ch options]
+    (when (not (instance? HttpChannel ch))
+      (throw (IllegalArgumentException. "Can't send map: channel is not an HTTP stream channel")))
+    (when (.sendStarted ^HttpChannel ch)
+      (throw (IllegalArgumentException. "Can't send map: this is not the first send to the channel")))
+    (dispatch-message (:body message) ch
+      (merge options (select-keys message [:status :headers]))))
+
   String
   (dispatch-message [message ch options]
     (wboss-send ch message options))
@@ -181,26 +190,28 @@
 (defn send!
   "Send a message to the channel, asynchronously.
 
-   `message` can either be a `String`, `File`, `InputStream`, `ISeq`,
-   or `byte[]`. If it is a `String`, it will be encoded to the character
-   set of the response for HTTP streams, and as UTF-8 for
-   WebSockets. `File`s and `InputStream`s will be sent as up to 16k
-   chunks (each chunk being a `byte[]` message for WebSockets). Each item
-   in an `ISeq` will pass through `send!`, and can be any of the valid
-   message types.
+  `message` can either be a `String`, `File`, `InputStream`, `ISeq`,
+  `byte[]`, or map. If it is a `String`, it will be encoded to the
+  character set of the response for HTTP streams, and as UTF-8 for
+  WebSockets. `File`s and `InputStream`s will be sent as up to 16k
+  chunks (each chunk being a `byte[]` message for WebSockets). Each
+  item in an `ISeq` will pass through `send!`, and can be any of the
+  valid message types.
+
+  If `message` is a map, its :body entry must be one of the other
+  valid message types, and its :status and :headers entries will be
+  used to override the status or headers returned from the handler
+  that called `as-channel` for HTTP streams. A map is *only* a valid
+  message on the first send to an HTTP stream channel - an exception
+  is thrown if it is passed on a subsequent send or passed to a
+  WebSocket channel.
 
-   The following options are supported in `options-map` [default]:
+  The following options are supported in `options-map` [default]:
 
    * :close? - if `true`, the channel will be closed when the send completes.
      Setting this to `true` on the first send to an HTTP stream channel
      will cause it to behave like a standard HTTP response, and *not* chunk
      the response. [false]
-   * :status - the HTTP status of the response. Used to override the status
-     returned from the handler that called `as-channel` for HTTP streams.
-     Ignored if this is not the first send to the channel. [nil]
-   * :headers - the HTTP headers of the response. Used to override the headers
-     returned from the handler that called `as-channel` for HTTP streams.
-     Ignored if this is not the first send to the channel. [nil]
    * :on-complete - `(fn [throwable] ...)` - called when the send
      attempt has completed. The success of the attempt is signaled by the
      passed value, i.e. if throwable is nil. If the error requires the
@@ -217,8 +228,7 @@
       u/kwargs-or-map->raw-map
       (o/validate-options send!))))
 
-(o/set-valid-options! send!
-  #{:close? :on-complete :status :headers})
+(o/set-valid-options! send! #{:close? :on-complete})
 
 (defn as-channel
   "Converts the current ring `request` in to an asynchronous channel.

web/test-integration/immutant/web/integ_test.clj

@@ -300,20 +300,52 @@
     (is request)
     (is (= "/" (:path-info request)))))
 
-(deftest send!-to-stream-with-status-header-overrides
+(deftest send!-to-stream-with-map-overrides-status-headers
   (replace-handler
     '(fn [request]
        (async/as-channel request
          :on-open (fn [ch]
-                    (async/send! ch "ahoy"
-                      {:close? true
-                       :status 201
-                       :headers {"foo" "bar"}})))))
+                    (async/send! ch {:body "ahoy"
+                                     :status 201
+                                     :headers {"foo" "bar"}}
+                      :close? true)))))
   (let [{:keys [body headers status]} (get-response (cdef-url))]
     (is (= "ahoy" body))
     (is (= 201 status))
     (is (= "bar" (:foo headers)))))
 
+(deftest send!-to-stream-with-map-after-send-has-started-throws
+  (replace-handler
+    '(do
+       (reset! client-state (promise))
+       (fn [request]
+         (async/as-channel request
+           :on-open (fn [ch]
+                      (async/send! ch "opening-")
+                      (try
+                        (async/send! ch {:body "ahoy"})
+                        (catch Exception e
+                          (deliver @client-state (.getMessage e))))
+                      (async/send! ch "closing" :close? true))))))
+  (is (= "opening-closing" (get-body (cdef-url))))
+  (is (re-find #"this is not the first send"
+        (read-string (get-body (str (cdef-url) "state"))))))
+
+(deftest send!-to-ws-with-map-throws
+  (replace-handler
+    '(do
+       (reset! client-state (promise))
+       (fn [request]
+         (async/as-channel request
+           :on-open (fn [ch]
+                      (try
+                        (async/send! ch {:body "ahoy"})
+                        (catch Exception e
+                          (deliver @client-state (.getMessage e)))))))))
+  (.close (ws/connect (cdef-url "ws")))
+  (is (re-find #"channel is not an HTTP stream channel"
+        (read-string (get-body (str (cdef-url) "state"))))))
+
 (deftest closing-a-stream-with-no-send-should-honor-original-response
   (replace-handler
     '(fn [request]