Merge pull request #394 from walmartlabs/hls/resolve-as-subscription-…

…support

Restore support for ResolverResult passed to subscription stream callback

deps.edn

@@ -51,5 +51,5 @@
    :exec-fn codox.main/generate-docs
    :exec-args {:metadata {:doc/format :markdown}
                :name "com.walmartlabs/lacinia"
-               :version "1.1-alpha-4"
+               :version "1.1-alpha-5a"
                :description "Clojure-native implementation of GraphQL"}}}}

dev-resources/subscriptions-schema.edn

@@ -7,5 +7,7 @@
  :subscriptions
  {:logs
   {:type :LogEvent
-   :args {:severity {:type String}}
+   :args {:severity {:type String}
+          :fakeError {:type Boolean
+                      :default false}}
    :stream :stream-logs}}}

docs/subscriptions/streamer.rst

@@ -78,4 +78,15 @@ are converted to responses in the response stream.
 Likewise, when the subscription is closed (by either the client or by the streamer itself),
 the callback will be invoked asynchronously.
 
+Notes
+-----
+
+The value passed to the source stream callback is normally a plain, non-nil value.
+
+It may be a wrapped value (e.g., via :api:`resolve/with-error`).
+This will be handled inside :api:`execute/execute-query` (which is invoked with the value passed to the callback).
+
+For historical reasons, it may also be a ResolverResult; it is the implementation's job to obtain
+the resolved value from this result before calling ``execute-query``; this is handled by lacinia-pedestal, for example.
+
 

project.clj

@@ -1,4 +1,4 @@
-(defproject com.walmartlabs/lacinia "1.1-alpha-4"
+(defproject com.walmartlabs/lacinia "1.1-alpha-5a"
   :description "A GraphQL server implementation in Clojure"
   :url "https://github.com/walmartlabs/lacinia"
   :license {:name "Apache, Version 2.0"

src/com/walmartlabs/lacinia/executor.clj

@@ -352,6 +352,14 @@
       :else
       (unwrap-resolver-result (invoke-resolver-for-field execution-context selection path container-type container-value)))))
 
+(defn ^:private unwrap-root-value
+  "For compatibility reasons, the value passed to a subscriber stream function may be a wrapped value."
+  [execution-context selection  value]
+  (if (su/is-wrapped-value? value)
+    (recur (su/apply-wrapped-value execution-context selection [(selection/alias-name selection)] value)
+           selection (:value value))
+    [execution-context value]))
+
 (defn execute-query
   "Entrypoint for execution of a query.
 
@@ -373,7 +381,7 @@
         context' (assoc context constants/schema-key schema)
         ;; Outside of subscriptions, the ::root-value is nil.
         ;; For subscriptions, the :root-value will be set to a non-nil value before
-        ;; executing the query.
+        ;; executing the query. It may be a wrapped value.
         root-type (get-nested parsed-query [:root :type-name])
         root-value (::resolved-value context)
         execution-context (map->ExecutionContext {:context context'
@@ -383,12 +391,13 @@
                                                   :*resolver-tracing *resolver-tracing
                                                   :timing-start timing-start
                                                   :*extensions *extensions})
+        [execution-context' root-value'] (unwrap-root-value execution-context (first selections) root-value)
         result-promise (resolve-promise)
         executor resolve/*callback-executor*
         f (bound-fn []
             (try
               (let [execute-fn (if (= :mutation operation-type) execute-nested-selections-sync execute-nested-selections)
-                    operation-result (execute-fn execution-context enabled-selections [] nil root-type root-value)]
+                    operation-result (execute-fn execution-context' enabled-selections [] nil root-type root-value')]
                 (resolve/on-deliver! operation-result
                   (fn [selected-data]
                     (let [errors (seq @*errors)

test/com/walmartlabs/lacinia/subscription_tests.clj

@@ -44,15 +44,21 @@
 
 (defn ^:private stream-logs
   [_context args source-stream]
-  (let [{:keys [severity]} args
+  (let [{:keys [severity fakeError]} args
         watch-key (gensym)]
     (add-watch *latest-log-event watch-key
                (fn [_ _ _ log-event]
                  (when *verbose*
                    (prn :log-event-received log-event))
-                 (when (or (nil? log-event)
-                           (nil? severity)
-                           (= severity (:severity log-event)))
+                 (cond
+                   (nil? log-event)
+                   (source-stream nil)
+
+                   fakeError
+                   (source-stream (resolve/resolve-as log-event {:message "Expected error"}))
+
+                   (or (nil? severity)
+                       (= severity (:severity log-event)))
                    (source-stream log-event))))
     #(remove-watch *latest-log-event watch-key)))
 
@@ -70,16 +76,23 @@
                            (parser/prepare-with-query-variables vars))
         *cleanup-callback (promise)
         context {constants/parsed-query-key prepared-query}
-        source-stream (fn [value]
-                        (if (some? value)
+        ;; For compatibility reasons, the value may be a ResolvedValue.
+        source-stream (fn accept-value [value]
+                        (cond
+                          (nil? value)
+                          (do
+                            (@*cleanup-callback)
+                            (reset! *latest-response nil))
+
+                          (resolve/is-resolver-result? value)
+                          (resolve/on-deliver! value accept-value)
+
+                          :else
                           (resolve/on-deliver!
                             (executor/execute-query (assoc context
-                                                      ::executor/resolved-value value))
+                                                           ::executor/resolved-value value))
                             (fn [result]
-                              (reset! *latest-response result)))
-                          (do
-                            (@*cleanup-callback)
-                            (reset! *latest-response nil))))]
+                              (reset! *latest-response result)))))]
     (deliver *cleanup-callback (executor/invoke-streamer context source-stream))))
 
 (deftest basic-subscription
@@ -139,13 +152,31 @@
     (is (= "Subscriptions only allow exactly one selection for the operation."
            (.getMessage e)))))
 
+(deftest errors-are-returned
+  (execute "subscription { logs(fakeError: true) { severity message } }" {})
+
+  (log-event {:severity "critical" :message "first"})
+
+  (is (= {:data {:logs {:message "first"
+                               :severity "critical"}}
+          :errors [{:message "Expected error"
+                    :locations [{:line 1, :column 16}]
+                    :path [:logs]
+                    :extensions {:arguments {:fakeError true}}}]}
+         (latest-response)))
+
+  (log-event nil)
+
+  (is (nil? (latest-response))))
 
 (deftest introspection
   (is (= {:data
           {:__schema
            {:subscriptionType
             {:description nil
-             :fields [{:args [{:name "severity"
+             :fields [{:args [{:name "fakeError"
+                               :type {:name "Boolean"}}
+                              {:name "severity"
                                :type {:name "String"}}]
                        :name "logs"
                        :type {:name "LogEvent"}}]}