   ([client dest data]
    (subscribe client dest data nil))
   ([client dest data {:keys [params] :or {params {}} :as opts}]
-   (let [[in out] (socket client dest data opts)
+   (let [[in out] (socket client dest data (assoc opts :_type :subscribe))
          bus (s/bus)]
      (s/subscribe in #(s/push! bus %) #(s/error! bus %) #(s/end! bus))
      (s/on-end bus #(s/end! out))

                        opts))))

;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;; EventSource
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;

(defn socket

   (socket client dest nil nil))
  ([client dest data]
   (socket client dest data nil))
  ([client dest data {:keys [params _type] :or {params {} _type :socket}}]
   (let [frame (encode {:data data :dest dest :type _type})
         req (prepare-request client frame :get params)
         uri (Uri. (:url req))]
     (.setQuery uri (:query-string req))
     (.setScheme uri (if (= (.getScheme uri) "http") "ws" "wss"))
     (let [ws (WebSocket. false)
           timer (Timer. 5000)
           busin (s/bus)
           streamin (s/filter #(not= (:type %) :ping) busin)
           busout (s/bus)]
       (letfn [(on-ws-message [event]
                 (let [frame (decode (.-message event))]
                   (s/push! busin frame)))
               (on-ws-error [event]
                 (s/error! busin event)
                 (.close ws))
               (on-ws-closed [event]
                 (s/end! busout)
                 (s/end! busin))
               (on-timer-tick [_]
                 (let [frame {:type :ping}]
                   (s/push! busout frame)))
