                                                                num-snapshots (inc num-items)))
              (and snapshot-beg
                   (= (:status r) :part)
-                  (= (:sub-status r) :snapshot-end)) (if-let [err (client-snapshot-end npr-client-loop client
+                  (= (:sub-status r) :snapshot-end)) (if-let [err (client-snapshot-end client
+                                                                                       npr-client-loop
                                                                                        stream-request
                                                                                        snapshot-beg r)]
                                                        (aput-close npr-client-loop out err)

  (:require-macros [cbfg.ago :refer [achan-buf ago ago-loop aclose
                                     aput aput-close atake]]))

(defn npr-client-stream-request-msg [actx client token])
(defn npr-snapshot-beg-msg [stream-request snapshot])
(defn npr-snapshot-end-msg [stream-request snapshot])
(defn npr-snapshot-item-msg [stream-request snapshot item])
(defn npr-rollback-msg [stream-request rollback])
(defn server-take-snapshot [actx server stream-request prev-snapshot])
(defn client-rollback [actx client stream-request rollback-msg])
(defn client-snapshot-beg [actx client stream-request snapshot-beg])
(defn client-snapshot-end [actx client stream-request snapshot-beg snapshot-end])
(defn client-snapshot-item [actx client stream-request snapshot-beg snapshot-item])

(defn make-npr-server-session [actx server stream-request to-client-ch]
  (ago-loop npr-server-session actx

                                                              (inc num-snapshots) num-items))
             (and (nil? snapshot-beg)
                  (= (:status r) :part)
                  (= (:sub-status r) :snapshot-beg)) (if-let [err (client-snapshot-beg client
                                                                                       npr-client-loop
                                                                                       stream-request r)]
                                                       (aput-close npr-client-loop out err)
                                                       (recur stream-request r
