;;   Copyright (c) 7theta. All rights reserved.
;;   The use and distribution terms for this software are covered by the
    MIT License (  ) which can also be
;;   found in the LICENSE file at the root of this distribution.
;;
;;   By using this software in any fashion, you are agreeing to be bound by
;;   the terms of this license.
;;   You must not remove this notice, or any others, from this software.

(ns via.endpoint
  (:refer-clojure :exclude [send])
  (:require [via.schema :as schema]
            [via.util.id :as id]
            [via.util.ns :as ns]
            [tempus.core :as t]
            #?(:clj [utilis.jar :as jar])
            #?(:clj [manifold.stream :as s]
               :cljs [vectio.util.manifold.stream :as s])
            #?(:clj [manifold.deferred :as d]
               :cljs [vectio.util.manifold.deferred :as d])
            #?(:clj [clojure.pprint :refer [pprint]]
               :cljs [cljs.pprint :refer [pprint]])
            [tempus.transit :as tt]
            [cognitect.transit :as transit]
            [utilis.timer :as ti]
            [signum.events :as se]
            [signum.subs :as ss]
            [signum.fx :as sfx]
            [integrant.core :as ig]
            [clojure.set :refer [map-invert rename-keys]])
  #?(:clj (:import [java.io ByteArrayInputStream ByteArrayOutputStream])))

(declare handle-handshake
         handle-handshake-reply
         handle-event
         handle-reply
         handle-error
         handshake-headers
         register deregister export
         message-encoder-key-map
         message-decoder-key-map
         message-type-encoder-key-map
         message-type-decoder-key-map
         id-generator
         shared-base-connection)

(defonce endpoints (atom {}))

(defn init
  ([] (init nil))
  ([{:keys [always-send-whole-values
            allow-deferred-subs
            exports
            transit-handlers
            streams
            on-error]
     :or {always-send-whole-values false
          allow-deferred-subs true}
     :as config}]
   (schema/validate-throw schema/EndpointConfig (into {} config))
   (let [endpoint (merge (when on-error {:on-error on-error})
                         {:id (id/uuid)
                          :connections (atom {})
                          :exports (atom {})
                          :always-send-whole-values (boolean always-send-whole-values)
                          :allow-deferred-subs (boolean allow-deferred-subs)
                          :deferred-connections (atom [])
                          :transit-handlers {:read {:handlers (merge (:read tt/handlers)
                                                                     (:read transit-handlers))}
                                             :write {:handlers (merge (:write tt/handlers)
                                                                      (:write transit-handlers))}}})]
     (doseq [[key exports] exports]
       #?(:cljs (when (= key :namespaces)
                  (js/console.error
                   "Namespace exported are available only in Clojure."
                   (pr-str {:exports exports}))))
       (doseq [id exports]
         (condp = key
           :events (export endpoint :event id)
           :subs (export endpoint :sub id)
           :namespaces #?(:clj (export endpoint :namespace id)
                          :cljs (do )))))
     (doseq [stream streams]
       (register endpoint stream))
     (swap! endpoints assoc (:id endpoint) endpoint)
     endpoint)))

(defn halt
  [{:keys [connections id] :as endpoint}]
  (doseq [[_ connection] @connections]
    (try (deregister endpoint connection)
         (catch #?(:clj Exception :cljs js/Error) e
           (handle-error
            endpoint
            :via.error/deregister-connection
            "Error occurred deregistering connection"
            {:connection connection
             :cause e}))))
  (swap! endpoints dissoc id)
  nil)

(defn handle-error
  [{:keys [on-error]} error-id message data]
  (locking #?(:clj Object :cljs js/Object)
    ((or on-error pprint)
     {:error-id error-id
      :message message
      :data data})))

(defn first-endpoint
  []
  (when (> (count @endpoints) 1)
    (throw (ex-info "Calling 'first-endpoint' is only available if there is at most 1 endpoint registered"
                    {:endpoints (count @endpoints)})))
  (first (vals @endpoints)))

(defmethod ig/init-key :via/endpoint
  [_ args]
  (init args))

(defmethod ig/halt-key! :via/endpoint
  [_ endpoint]
  (halt endpoint))

;;; Messaging

(defn encode*
  [endpoint message]
  (let [handlers (-> endpoint :transit-handlers :write)]
    #?(:clj (let [out (ByteArrayOutputStream. 4096)]
              (transit/write (transit/writer out :json handlers) message)
              (.toString out))
       :cljs (transit/write (transit/writer :json handlers) message))))

(defn decode*
  [endpoint ^String message]
  (let [handlers (-> endpoint :transit-handlers :read)]
    #?(:clj (let [in (ByteArrayInputStream. (.getBytes message))]
              (transit/read (transit/reader in :json handlers)))
       :cljs (transit/read (transit/reader :json handlers) message))))

(defn encode
  [endpoint message]
  (->> (-> message
           (update :type message-type-encoder-key-map)
           (rename-keys message-encoder-key-map))
       (encode* endpoint)))

(defn decode
  [endpoint message]
  (-> endpoint
      (decode* message)
      (rename-keys message-decoder-key-map)
      (update :type message-type-decoder-key-map)))

(defn handle-message
  [endpoint {:keys [trace-log] :as connection} encoded]
  (when-let [decoded (try (->> encoded
                               (decode endpoint)
                               (schema/validate-throw schema/Message))
                          (catch #?(:clj Exception :cljs js/Error) e
                            (handle-error
                             endpoint :via.error/handle-message
                             "Exception occurred decoding message"
                             {:cause e
                              :message encoded})
                            nil))]
    (try
      (when trace-log
        (swap! trace-log conj
               {:message decoded
                :timestamp (t/now)}))
      (condp = (:type decoded)
        :event (handle-event endpoint connection decoded)
        :reply (handle-reply endpoint connection decoded)
        :handshake (handle-handshake endpoint connection decoded)
        :handshake-reply (handle-handshake-reply endpoint connection decoded)
        (handle-error
         endpoint :via.error/malformed-message
         "Malformed message"
         {:message decoded}))
      (catch #?(:clj Exception :cljs js/Error) e
        (handle-error
         endpoint :via.error/handle-message
         "Exception occurred handling message"
         {:message decoded
          :cause e})))))

(defn send
  [endpoint {:keys [underlying-stream trace-log]} message]
  (when trace-log
    (swap! trace-log conj
           {:message message
            :timestamp (t/now)}))
  (s/put! underlying-stream (encode endpoint message)))

(defn send-all
  [endpoint message]
  (doseq [connection (vals @(:connections endpoint))]
    (send endpoint connection message)))

(defn remove-request-handler
  [connection request-id]
  (swap! (:requests connection) dissoc request-id))

(defn add-request-handler
  [connection {:keys [timeout on-error] :as opts}]
  (let [request-id ((:next-id connection))
        timer (when timeout
                (ti/run-after
                 #(do (remove-request-handler connection request-id)
                      (when on-error (on-error)))
                 timeout))]
    (swap! (:requests connection) assoc request-id
           (assoc opts :timer timer))
    request-id))

(defn send-reply
  [endpoint connection original-message reply]
  (let [{:keys [request-id]} original-message]
    (when request-id
      (send endpoint connection
            {:body reply
             :type :reply
             :request-id request-id}))))


  Connection Management

(defn connection
  [endpoint connection-id]
  (get @(:connections endpoint) connection-id))

(defn lookup
  [endpoint found?]
  (some (fn [connection]
          (when (found? connection)
            connection))
        (vals @(:connections endpoint))))

(defn lookup-stream
  [endpoint stream]
  (lookup endpoint (comp (partial = stream) :underlying-stream)))

(defn deregister
  [endpoint {:keys [id via-stream on-deregister ready subs]}]
  (locking (:connections endpoint)
    (when (get @(:connections endpoint) id)
      (reset! ready false)
      (swap! (:connections endpoint) dissoc id)
      (s/close! via-stream)
      (when on-deregister (on-deregister))
      (doseq [[_ cleanup] (get @(:remote-dispose subs) id)]
        (cleanup))
      true)))

(defn deferred-connection
  [{:keys [deferred-connections] :as endpoint}]
  (or (first @deferred-connections)
      (let [connection (merge {:id (id/uuid)
                               :endpoint (constantly endpoint)
                               :handshake-completed (atom false)}
                              (shared-base-connection))]
        (swap! deferred-connections conj connection)
        connection)))

(defn handshake
  [{:keys [endpoint] :as connection}]
  (send endpoint connection
        {:type :handshake
         :headers (handshake-headers)}))

(defn register
  ([endpoint stream] (register endpoint stream nil))
  ([endpoint stream additional-params]
   (if (d/deferred? stream)
     (let [connection (deferred-connection endpoint)]
       (d/on-realized stream
                      #(register endpoint % additional-params)
                      #(throw %))
       connection)
     (locking (:connections endpoint)
       (when (lookup-stream endpoint stream)
         (throw (ex-info "Stream is already registered" {:stream stream})))
       (let [additional-params (merge {:upstream? true} additional-params)
             connection (when-let [connection (first @(:deferred-connections endpoint))]
                          (swap! (:deferred-connections endpoint)
                                 #(->> %
                                       (remove
                                        (comp (partial = (:id connection))
                                           :id))
                                       vec))
                          connection)
             id (or (:id connection) (id/uuid))
             via-stream (s/stream)
             connection (cond->> (merge {:id id}
                                        (when (not connection)
                                          (shared-base-connection))
                                        (when (:trace additional-params)
                                          {:trace-log (atom [])})
                                        connection
                                        {:requests (atom {})
                                         :underlying-stream stream
                                         :via-stream via-stream
                                         :endpoint (constantly endpoint)})
                          additional-params (merge additional-params))]
         (swap! (:connections endpoint) assoc id connection)
         (s/connect stream via-stream)
         (s/on-closed via-stream (fn []
                                   (deregister endpoint connection)
                                   (when (:upstream? additional-params)
                                     (s/close! stream))))
         (s/on-closed stream #(when (not (s/closed? via-stream))
                                (s/close! via-stream)))
         (s/consume #(handle-message endpoint connection %) via-stream)
         (handshake connection)
         connection)))))

(defn first-connection
  [{:keys [connections]}]
  (when (> (count @connections) 1)
    (throw (ex-info "Calling 'first-connection' is only available if there is at most 1 connection available"
                    {:connections (count @connections)})))
  (first (vals @connections)))

(defn reconnecting-stream
  ([endpoint underlying-stream]
   (reconnecting-stream endpoint underlying-stream nil))
  ([endpoint underlying-stream {:keys [on-connect
                                       on-reconnect
                                       on-disconnect
                                       params]}]
   (let [via-conn (atom nil)]
     (s/consume
      (fn [{:keys [stream initial error]}]
        (cond
          (and stream initial)
          (let [connection (->> params
                                (register endpoint stream)
                                (reset! via-conn))]
            (when on-connect
              (on-connect connection)))

          (and stream (not initial))
          (do (handshake @via-conn)
              (when on-reconnect (on-reconnect)))

          error
          (do (reset! (:ready @via-conn) false)
              (when on-disconnect (on-disconnect error)))))
      underlying-stream)
     underlying-stream)))

;;; Exports

(defn export
  [endpoint type id]
  (schema/validate-throw schema/ExportType type)
  (swap! (:exports endpoint) update type
         #(conj (set %) id)))

(defn unexport
  [endpoint type id]
  (schema/validate-throw schema/ExportType type)
  (swap! (:exports endpoint) update type
         #(disj (set %) id)))

(defn exported?
  [endpoint type id]
  (schema/validate-throw schema/ExportType type)
  (boolean
   (or (get-in @(:exports endpoint) [type id])
       (when (#{:sub :event} type)
         (some #(= (ns/normalize-namespace %)
                   (ns/normalize-namespace
                    (if (= :event type)
                      (se/namespace id)
                      (ss/namespace id))))
               (:namespace @(:exports endpoint)))))))


;;; Effect Handlers

(sfx/reg-fx
 :via/disconnect
 (fn [{:keys [connection endpoint]} _data]
   (deregister endpoint connection)))

(sfx/reg-fx
 :via/reply
 (fn [{:keys [message endpoint connection]} data]
   (send-reply endpoint connection message data)))

(sfx/reg-fx
 :via/send
 (fn [{:keys [endpoint]} {:keys [connection-id message]}]
   (schema/validate-some connection-id)
   (schema/validate-throw schema/QueryVector message)
   (let [connection (connection endpoint connection-id)]
     (schema/validate-some connection)
     (send endpoint connection
           {:body message
            :type :event}))))


;;; Private

(defn- handshake-headers
  []
  {:via-language #?(:clj :clj :cljs :cljs)
   :via-version #?(:clj (jar/version "com.7theta" "via")
                   :cljs nil)})

(defn- handle-handshake
  [endpoint {:keys [on-handshake] :as connection} message]
  (send endpoint connection
        {:type :handshake-reply
         :headers (handshake-headers)})
  (when on-handshake (on-handshake message)))

(defn- handle-handshake-reply
  [_endpoint {:keys [ready on-handshake-reply]} message]
  (reset! ready true)
  (when on-handshake-reply (on-handshake-reply message)))

(defn- handle-reply
  [_endpoint connection message]
  (let [{:keys [body request-id]} message]
    (when-let [{:keys [on-reply timer]} (and request-id (get @(:requests connection) request-id))]
      (when timer (ti/cancel timer))
      (remove-request-handler connection request-id)
      (on-reply body))))

(defn- handle-event
  [endpoint {:keys [coeffects] :as connection} message]
  (let [{:keys [body]} message
        [event-id & _] body]
    (if (and (se/event? event-id)
             (exported? endpoint :event event-id))
      (se/dispatch (merge coeffects
                          {:message message
                           :endpoint endpoint
                           :connection connection})
                   body)
      (let [error-text "Attempted to handle an event that either does not exist or has not been exported."]
        (handle-error
         endpoint
         :via.error/unknown-or-non-exported-event
         error-text
         {:message message})
        (send-reply endpoint connection message
                    {:status 400
                     :body error-text})))))

(def message-encoder-key-map
  {:body :b
   :headers :h
   :type :t
   :request-id :r})

(def message-decoder-key-map
  (map-invert message-encoder-key-map))

(def message-type-encoder-key-map
  {:event 0
   :reply 1
   :handshake 2
   :handshake-reply 3})

(def message-type-decoder-key-map
  (map-invert message-type-encoder-key-map))

(defn id-generator
  []
  (let [last-id (atom 0)]
    #(swap! last-id inc)))

(defn shared-base-connection
  []
  {:ready (atom false)
   :next-id (id-generator)
   :subs {:signals (atom {})
          :remote-dispose (atom {})}})

   Copyright (c) 7theta. All rights reserved.
   The use and distribution terms for this software are covered by the
   found in the LICENSE file at the root of this distribution.

   By using this software in any fashion, you are agreeing to be bound by
   the terms of this license.
   You must not remove this notice, or any others, from this software.
 Messaging
 Exports
 Effect Handlers
 Private
    MIT License (  ) which can also be

(ns via.endpoint
  (:refer-clojure :exclude [send])
  (:require [via.schema :as schema]
            [via.util.id :as id]
            [via.util.ns :as ns]
            [tempus.core :as t]
            #?(:clj [utilis.jar :as jar])
            #?(:clj [manifold.stream :as s]
               :cljs [vectio.util.manifold.stream :as s])
            #?(:clj [manifold.deferred :as d]
               :cljs [vectio.util.manifold.deferred :as d])
            #?(:clj [clojure.pprint :refer [pprint]]
               :cljs [cljs.pprint :refer [pprint]])
            [tempus.transit :as tt]
            [cognitect.transit :as transit]
            [utilis.timer :as ti]
            [signum.events :as se]
            [signum.subs :as ss]
            [signum.fx :as sfx]
            [integrant.core :as ig]
            [clojure.set :refer [map-invert rename-keys]])
  #?(:clj (:import [java.io ByteArrayInputStream ByteArrayOutputStream])))

(declare handle-handshake
         handle-handshake-reply
         handle-event
         handle-reply
         handle-error
         handshake-headers
         register deregister export
         message-encoder-key-map
         message-decoder-key-map
         message-type-encoder-key-map
         message-type-decoder-key-map
         id-generator
         shared-base-connection)

(defonce endpoints (atom {}))

(defn init
  ([] (init nil))
  ([{:keys [always-send-whole-values
            allow-deferred-subs
            exports
            transit-handlers
            streams
            on-error]
     :or {always-send-whole-values false
          allow-deferred-subs true}
     :as config}]
   (schema/validate-throw schema/EndpointConfig (into {} config))
   (let [endpoint (merge (when on-error {:on-error on-error})
                         {:id (id/uuid)
                          :connections (atom {})
                          :exports (atom {})
                          :always-send-whole-values (boolean always-send-whole-values)
                          :allow-deferred-subs (boolean allow-deferred-subs)
                          :deferred-connections (atom [])
                          :transit-handlers {:read {:handlers (merge (:read tt/handlers)
                                                                     (:read transit-handlers))}
                                             :write {:handlers (merge (:write tt/handlers)
                                                                      (:write transit-handlers))}}})]
     (doseq [[key exports] exports]
       #?(:cljs (when (= key :namespaces)
                  (js/console.error
                   "Namespace exported are available only in Clojure."
                   (pr-str {:exports exports}))))
       (doseq [id exports]
         (condp = key
           :events (export endpoint :event id)
           :subs (export endpoint :sub id)
           :namespaces #?(:clj (export endpoint :namespace id)
                          :cljs (do )))))
     (doseq [stream streams]
       (register endpoint stream))
     (swap! endpoints assoc (:id endpoint) endpoint)
     endpoint)))

(defn halt
  [{:keys [connections id] :as endpoint}]
  (doseq [[_ connection] @connections]
    (try (deregister endpoint connection)
         (catch #?(:clj Exception :cljs js/Error) e
           (handle-error
            endpoint
            :via.error/deregister-connection
            "Error occurred deregistering connection"
            {:connection connection
             :cause e}))))
  (swap! endpoints dissoc id)
  nil)

(defn handle-error
  [{:keys [on-error]} error-id message data]
  (locking #?(:clj Object :cljs js/Object)
    ((or on-error pprint)
     {:error-id error-id
      :message message
      :data data})))

(defn first-endpoint
  []
  (when (> (count @endpoints) 1)
    (throw (ex-info "Calling 'first-endpoint' is only available if there is at most 1 endpoint registered"
                    {:endpoints (count @endpoints)})))
  (first (vals @endpoints)))

(defmethod ig/init-key :via/endpoint
  [_ args]
  (init args))

(defmethod ig/halt-key! :via/endpoint
  [_ endpoint]
  (halt endpoint))


(defn encode*
  [endpoint message]
  (let [handlers (-> endpoint :transit-handlers :write)]
    #?(:clj (let [out (ByteArrayOutputStream. 4096)]
              (transit/write (transit/writer out :json handlers) message)
              (.toString out))
       :cljs (transit/write (transit/writer :json handlers) message))))

(defn decode*
  [endpoint ^String message]
  (let [handlers (-> endpoint :transit-handlers :read)]
    #?(:clj (let [in (ByteArrayInputStream. (.getBytes message))]
              (transit/read (transit/reader in :json handlers)))
       :cljs (transit/read (transit/reader :json handlers) message))))

(defn encode
  [endpoint message]
  (->> (-> message
           (update :type message-type-encoder-key-map)
           (rename-keys message-encoder-key-map))
       (encode* endpoint)))

(defn decode
  [endpoint message]
  (-> endpoint
      (decode* message)
      (rename-keys message-decoder-key-map)
      (update :type message-type-decoder-key-map)))

(defn handle-message
  [endpoint {:keys [trace-log] :as connection} encoded]
  (when-let [decoded (try (->> encoded
                               (decode endpoint)
                               (schema/validate-throw schema/Message))
                          (catch #?(:clj Exception :cljs js/Error) e
                            (handle-error
                             endpoint :via.error/handle-message
                             "Exception occurred decoding message"
                             {:cause e
                              :message encoded})
                            nil))]
    (try
      (when trace-log
        (swap! trace-log conj
               {:message decoded
                :timestamp (t/now)}))
      (condp = (:type decoded)
        :event (handle-event endpoint connection decoded)
        :reply (handle-reply endpoint connection decoded)
        :handshake (handle-handshake endpoint connection decoded)
        :handshake-reply (handle-handshake-reply endpoint connection decoded)
        (handle-error
         endpoint :via.error/malformed-message
         "Malformed message"
         {:message decoded}))
      (catch #?(:clj Exception :cljs js/Error) e
        (handle-error
         endpoint :via.error/handle-message
         "Exception occurred handling message"
         {:message decoded
          :cause e})))))

(defn send
  [endpoint {:keys [underlying-stream trace-log]} message]
  (when trace-log
    (swap! trace-log conj
           {:message message
            :timestamp (t/now)}))
  (s/put! underlying-stream (encode endpoint message)))

(defn send-all
  [endpoint message]
  (doseq [connection (vals @(:connections endpoint))]
    (send endpoint connection message)))

(defn remove-request-handler
  [connection request-id]
  (swap! (:requests connection) dissoc request-id))

(defn add-request-handler
  [connection {:keys [timeout on-error] :as opts}]
  (let [request-id ((:next-id connection))
        timer (when timeout
                (ti/run-after
                 #(do (remove-request-handler connection request-id)
                      (when on-error (on-error)))
                 timeout))]
    (swap! (:requests connection) assoc request-id
           (assoc opts :timer timer))
    request-id))

(defn send-reply
  [endpoint connection original-message reply]
  (let [{:keys [request-id]} original-message]
    (when request-id
      (send endpoint connection
            {:body reply
             :type :reply
             :request-id request-id}))))


  Connection Management

(defn connection
  [endpoint connection-id]
  (get @(:connections endpoint) connection-id))

(defn lookup
  [endpoint found?]
  (some (fn [connection]
          (when (found? connection)
            connection))
        (vals @(:connections endpoint))))

(defn lookup-stream
  [endpoint stream]
  (lookup endpoint (comp (partial = stream) :underlying-stream)))

(defn deregister
  [endpoint {:keys [id via-stream on-deregister ready subs]}]
  (locking (:connections endpoint)
    (when (get @(:connections endpoint) id)
      (reset! ready false)
      (swap! (:connections endpoint) dissoc id)
      (s/close! via-stream)
      (when on-deregister (on-deregister))
      (doseq [[_ cleanup] (get @(:remote-dispose subs) id)]
        (cleanup))
      true)))

(defn deferred-connection
  [{:keys [deferred-connections] :as endpoint}]
  (or (first @deferred-connections)
      (let [connection (merge {:id (id/uuid)
                               :endpoint (constantly endpoint)
                               :handshake-completed (atom false)}
                              (shared-base-connection))]
        (swap! deferred-connections conj connection)
        connection)))

(defn handshake
  [{:keys [endpoint] :as connection}]
  (send endpoint connection
        {:type :handshake
         :headers (handshake-headers)}))

(defn register
  ([endpoint stream] (register endpoint stream nil))
  ([endpoint stream additional-params]
   (if (d/deferred? stream)
     (let [connection (deferred-connection endpoint)]
       (d/on-realized stream
                      #(register endpoint % additional-params)
                      #(throw %))
       connection)
     (locking (:connections endpoint)
       (when (lookup-stream endpoint stream)
         (throw (ex-info "Stream is already registered" {:stream stream})))
       (let [additional-params (merge {:upstream? true} additional-params)
             connection (when-let [connection (first @(:deferred-connections endpoint))]
                          (swap! (:deferred-connections endpoint)
                                 #(->> %
                                       (remove
                                        (comp (partial = (:id connection))
                                           :id))
                                       vec))
                          connection)
             id (or (:id connection) (id/uuid))
             via-stream (s/stream)
             connection (cond->> (merge {:id id}
                                        (when (not connection)
                                          (shared-base-connection))
                                        (when (:trace additional-params)
                                          {:trace-log (atom [])})
                                        connection
                                        {:requests (atom {})
                                         :underlying-stream stream
                                         :via-stream via-stream
                                         :endpoint (constantly endpoint)})
                          additional-params (merge additional-params))]
         (swap! (:connections endpoint) assoc id connection)
         (s/connect stream via-stream)
         (s/on-closed via-stream (fn []
                                   (deregister endpoint connection)
                                   (when (:upstream? additional-params)
                                     (s/close! stream))))
         (s/on-closed stream #(when (not (s/closed? via-stream))
                                (s/close! via-stream)))
         (s/consume #(handle-message endpoint connection %) via-stream)
         (handshake connection)
         connection)))))

(defn first-connection
  [{:keys [connections]}]
  (when (> (count @connections) 1)
    (throw (ex-info "Calling 'first-connection' is only available if there is at most 1 connection available"
                    {:connections (count @connections)})))
  (first (vals @connections)))

(defn reconnecting-stream
  ([endpoint underlying-stream]
   (reconnecting-stream endpoint underlying-stream nil))
  ([endpoint underlying-stream {:keys [on-connect
                                       on-reconnect
                                       on-disconnect
                                       params]}]
   (let [via-conn (atom nil)]
     (s/consume
      (fn [{:keys [stream initial error]}]
        (cond
          (and stream initial)
          (let [connection (->> params
                                (register endpoint stream)
                                (reset! via-conn))]
            (when on-connect
              (on-connect connection)))

          (and stream (not initial))
          (do (handshake @via-conn)
              (when on-reconnect (on-reconnect)))

          error
          (do (reset! (:ready @via-conn) false)
              (when on-disconnect (on-disconnect error)))))
      underlying-stream)
     underlying-stream)))


(defn export
  [endpoint type id]
  (schema/validate-throw schema/ExportType type)
  (swap! (:exports endpoint) update type
         #(conj (set %) id)))

(defn unexport
  [endpoint type id]
  (schema/validate-throw schema/ExportType type)
  (swap! (:exports endpoint) update type
         #(disj (set %) id)))

(defn exported?
  [endpoint type id]
  (schema/validate-throw schema/ExportType type)
  (boolean
   (or (get-in @(:exports endpoint) [type id])
       (when (#{:sub :event} type)
         (some #(= (ns/normalize-namespace %)
                   (ns/normalize-namespace
                    (if (= :event type)
                      (se/namespace id)
                      (ss/namespace id))))
               (:namespace @(:exports endpoint)))))))



(sfx/reg-fx
 :via/disconnect
 (fn [{:keys [connection endpoint]} _data]
   (deregister endpoint connection)))

(sfx/reg-fx
 :via/reply
 (fn [{:keys [message endpoint connection]} data]
   (send-reply endpoint connection message data)))

(sfx/reg-fx
 :via/send
 (fn [{:keys [endpoint]} {:keys [connection-id message]}]
   (schema/validate-some connection-id)
   (schema/validate-throw schema/QueryVector message)
   (let [connection (connection endpoint connection-id)]
     (schema/validate-some connection)
     (send endpoint connection
           {:body message
            :type :event}))))



(defn- handshake-headers
  []
  {:via-language #?(:clj :clj :cljs :cljs)
   :via-version #?(:clj (jar/version "com.7theta" "via")
                   :cljs nil)})

(defn- handle-handshake
  [endpoint {:keys [on-handshake] :as connection} message]
  (send endpoint connection
        {:type :handshake-reply
         :headers (handshake-headers)})
  (when on-handshake (on-handshake message)))

(defn- handle-handshake-reply
  [_endpoint {:keys [ready on-handshake-reply]} message]
  (reset! ready true)
  (when on-handshake-reply (on-handshake-reply message)))

(defn- handle-reply
  [_endpoint connection message]
  (let [{:keys [body request-id]} message]
    (when-let [{:keys [on-reply timer]} (and request-id (get @(:requests connection) request-id))]
      (when timer (ti/cancel timer))
      (remove-request-handler connection request-id)
      (on-reply body))))

(defn- handle-event
  [endpoint {:keys [coeffects] :as connection} message]
  (let [{:keys [body]} message
        [event-id & _] body]
    (if (and (se/event? event-id)
             (exported? endpoint :event event-id))
      (se/dispatch (merge coeffects
                          {:message message
                           :endpoint endpoint
                           :connection connection})
                   body)
      (let [error-text "Attempted to handle an event that either does not exist or has not been exported."]
        (handle-error
         endpoint
         :via.error/unknown-or-non-exported-event
         error-text
         {:message message})
        (send-reply endpoint connection message
                    {:status 400
                     :body error-text})))))

(def message-encoder-key-map
  {:body :b
   :headers :h
   :type :t
   :request-id :r})

(def message-decoder-key-map
  (map-invert message-encoder-key-map))

(def message-type-encoder-key-map
  {:event 0
   :reply 1
   :handshake 2
   :handshake-reply 3})

(def message-type-decoder-key-map
  (map-invert message-type-encoder-key-map))

(defn id-generator
  []
  (let [last-id (atom 0)]
    #(swap! last-id inc)))

(defn shared-base-connection
  []
  {:ready (atom false)
   :next-id (id-generator)
   :subs {:signals (atom {})
          :remote-dispose (atom {})}})
