This repository has been archived by the owner on Mar 22, 2023. It is now read-only.
forked from mpenet/jet
-
Notifications
You must be signed in to change notification settings - Fork 4
/
websocket.clj
92 lines (73 loc) · 3.43 KB
/
websocket.clj
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
(ns qbits.jet.client.websocket
(:require
[clojure.core.async :as async]
[qbits.jet.websocket :as ws])
(:import
(java.net URI)
(org.eclipse.jetty.websocket.client ClientUpgradeRequest)))
(defrecord Connection [client request socket])
(defn connect!
"Takes a WebSocketClient, a url, a handler, an option map and returns a websocket
client.
The option map can take the following keys:
* `:executor` - java.util.concurrent.Executor instance
* `:ssl-context-factory` - SSLContextFactory instance to be use to perform wss
requests http://download.eclipse.org/jetty/stable-9/apidocs/org/eclipse/jetty/util/ssl/SslContextFactory.html
* `:in` - fn that returns the :in channel that the handler will receive -
defaults to c.core.async/chan
* `:out` - fn that returns the :in channel that the handler will receive -
defaults to c.core.async/chan
* `:ctrl` - fn that returns the :in channel that the handler will receive -
defaults to c.core.async/chan
The handler receives a map of:
* `:in` - core.async chan that receives data sent by the client
* `:out` - core async chan you can use to send data to client, or close the
connection by closing the channel
* `:ctrl` - core.asyn chan that received control messages such as:
``[::error e]`, `[::close reason]`
* `:ws` - qbits.jet.websocket/WebSocket instance"
[client url handler & [{:as options
:keys [in out ctrl
executor
ssl-context-factory
async-write-timeout
connect-timeout
max-idle-timeout
max-binary-message-buffer-size
max-text-message-buffer-size
subprotocols
daemon?
middleware]
:or {in async/chan
out async/chan
ctrl async/chan}}]]
(let [request (ClientUpgradeRequest.)
ws (ws/make-websocket (in) (out) (ctrl) handler)]
(when subprotocols
(.setSubProtocols ^ClientUpgradeRequest request
^"[Ljava.lang.String;" (into-array String subprotocols)))
(when executor
(.setExecutor client executor))
(when max-idle-timeout
(.setMaxIdleTimeout client (long max-idle-timeout)))
;; async timeout has an assertion related with idle timeout, it must be set after setting max idle timeout
(when async-write-timeout
(.setAsyncWriteTimeout client (long async-write-timeout)))
(when connect-timeout
(.setConnectTimeout client (long connect-timeout)))
(when max-binary-message-buffer-size
(.setMaxBinaryMessageBufferSize client (int max-binary-message-buffer-size)))
(when max-text-message-buffer-size
(.setMaxTextMessageBufferSize client
(int max-text-message-buffer-size)))
;; (.setDaemon client daemon?)
;; void setBindAdddress(SocketAddress bindAddress)
;; void setCookieStore(CookieStore cookieStore)
;; void setEventDriverFactory(EventDriverFactory factory)
;; void setMasker(Masker masker)
;; void setSessionFactory(SessionFactory sessionFactory)
(when (fn? middleware)
(middleware client request))
(.start client)
(.connect client ws (URI/create url) request)
(Connection. client request ws)))