                    (.send ws data)))
                (on-busout-end []
                  (.close ws))]
+
          (s/on-end busout on-busout-end)
          (s/on-value busout on-busout-value)
 
          (events/listen ws EventType.MESSAGE on-ws-message)
          (events/listen ws EventType.ERROR on-ws-error)
+         (events/listen ws EventType.CLOSED on-ws-closed)
          (events/listen timer Timer.TICK on-timer-tick)
+
+         (.start timer)
+         (.open ws (.toString uri))
 
          [streamin busout])))))
 

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

                        opts))))

;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;; Socket
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;

(defn socket
