                                                                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)

             (= (:status r) :ok) (aput-close npr-client-loop out {:status :ok})
             (and snapshot-beg
                  (= (:status r) :part)
                  (= (:sub-status r) :snapshot-item)) (if-let [err (client-snapshot-item npr-client-loop client
                                                                                         stream-request
                                                                                         snapshot-beg r)]
                                                        (aput-close npr-client-loop out err)

                                                              num-snapshots num-items))
             (and (nil? snapshot-beg)
                  (= (:status :rollback))) (aput-close npr-client-loop out
                                                       (client-rollback client
                                                                        npr-client-loop
                                                                        stream-request r))
             :else (aput-close npr-client-loop out {:status :unexpected-msg}))))

(defn make-npr-client-session [actx client token to-server-ch from-server-ch]
  (let [out (achan-buf actx 1)
        stream-request (client-stream-request-msg client actx token)]
    (ago npr-client-session actx
         (aput npr-client-session to-server-ch stream-request)
         (npr-client-loop npr-client-session client stream-request
