Recently I have been exploring core.async and websockets. In this tutorial I will demonstrate how to build a chatroom using clojurescript, core.async and websockets.
We will be using Chord, core.async, http-kit and reagent.
core.async fundamentals
Interacting with channels.
core.async is a library for Clojure/Clojurescript that implements communicating sequential processes.
Core.async provides mechanisms to manage asynchronous tasks using lightweight threads called “go-blocks” and queue-like channels. Using core.async will let us write elegant code that reads synchronously, and avoids the use of callbacks.
go-blocks need a way to pass data between each other. This is done by interacting with channels.
By default a channel is unbuffered. This means that if the channel contains an item, no further items can be written to the channel, until the channel is read from. If a channel is full, go-routines that try to write more data to the channel will “park”. This corresponds to “blocking” in a threaded environment. Buffered channels can easily be created to allow a certain amount of items to buffer. When the buffer is full, attempts to add an additional item to the channel will result in the go-routine parking.
(def mychan (chan)) ; Unbuffered channel
(def buffered-chan (chan 5)) ; channel with buffer of size '5'
; Use '!>' to put an item on a channel within a go-routine
; Use '!<' to take an item on a channel within a go-routine
(go
(!> my-chan "hello")
(!> my-chan "world")) ; This go-block will park
; Print hello world infinitely
(defn loop-1
[]
(go-loop
(!> my-chan "hello")
(!> my-chan "world")
(recur)))
(defn loop-2
[]
(go-loop
(println (!< my-chan)
(recur)))
(loop-1)
(loop-2)
Handling concurrency with alt! and mult.
Core.async provides a killer feature with the alt! function, which allows your program to park on multiple channels, similar to select() from C/POSIX programming.
The alt! function accepts a sequence of clauses. Each clause consists of a channel to take from, and a result expression to run when there is data in the channel.
(alt!
my-chan ([message]
(println "Message from unbuffered-chan" message))
buffered-chan ([message)]
(println "Message from buffered-chan" message))
In the above example when alt! is invoked, it will park until one of the channels has data available.
Using mult and tap we can create a multiplexer from the channel we want multiple clients to read from. With this multiplexer, we can then create taps, from the multiplexer to new channels. This allows each client to have its own channel to read from, without read-race contention. A good pattern to follow is to have all the clients write to the same channel, and and all the clients read from their own private tapped channels.
(def main-channel (chan))
(def main-multiplexer (mult main-channel))
(def reader-channel (chan))
(tap main-multiplexer reader-channel)
(go
(>! main-channel "Hello World")
(println (<! reader-channel)))
Wiring up our chat room
To implement our chatroom we will use Chord to create a core.async channel abstractions on top of websockets.
Chord provides a single macro, with-channel. This macro gives us a channel for the websocket that is setup when a client connects to our websocket endpoint. We setup a main-channel, from which all connected clients will receive messages through a tap that is setup for each connected client. When we receive a message on the client’s websocket, we will read the message from the websocket channel and write it to our main chatroom channel. This will result in all the client’s receiving the new message on their tapped channels. When the taps receive a channel, we read from the tapped channel, and write the message to the websocket channel.
Since we want all messages to have a unique id, we will create the main channel with a transducer. The transducer will modify every new message on the channel with the mapping function.
; Server code
(ns hablamos.handler
(:require
[org.httpkit.server :as hk]
[chord.http-kit :refer [with-channel]]
[compojure.core :refer :all]
[compojure.route :as route]
[clojure.core.async :as a]
[medley.core :refer [random-uuid]]))
; Use a transducer to append a unique id to each message
; To use a transducer on a channel, you must specify the channel buffer size
(defonce main-chan (a/chan 1 (map #(assoc % :id (random-uuid)))))
(defonce main-mult (a/mult main-chan))
(def users (atom {}))
(defn ws-handler
[req]
(with-channel req ws-ch
(let [client-tap (a/chan)
client-id (random-uuid)]
(a/tap main-mult client-tap)
(a/go-loop []
(a/alt!
client-tap ([message]
(if message
(do
(a/>! ws-ch message)
(recur))
(a/close! ws-ch)))
ws-ch ([{:keys [message]}]
(if message
(let [{:keys [msg m-type]} message]
(do
(when (= m-type :new-user)
(swap! users assoc client-id msg)
(a/>! ws-ch {:id (random-uuid)
:msg (set (vals @users))
:m-type :init-users}))
(a/>! main-chan message)
(recur)))
(do
(a/untap main-mult client-tap)
(a/>! main-chan {:m-type :user-left
:msg (get @users client-id)})
(swap! users dissoc client-id)))))))))
(defroutes app
(GET "/ws" [] ws-handler))
On the client side our code is simpler, because we only need to handle establishing a websocket connection, sending new messages to the server, and reading from our websocket channel to show new messages to the user.
Chord only provides one function for clojurescript, chord.client/ws-ch. This function establishes a websocket and abstracts it as a core.async channel.
(ns hablamos.core
(:require [reagent.core :as reagent :refer [atom]]
[chord.client :refer [ws-ch]]
[cljs.core.async :as a :refer [>! <! put!]])
(:require-macros [cljs.core.async.macros :refer [go go-loop]]))
(defonce msg-list (atom []))
(defonce users-set (atom #{}))
(defonce send-chan (a/chan))
(defn setup-websockets! []
(go
(let [{:keys [ws-channel error]} (<! (ws-ch "ws://localhost:3449/ws"))]
(if error
(println "Something went wrong with the websocket")
(do
(send-msg {:m-type :new-user
:msg (:user @app-state)})
(send-msgs ws-channel)
(receive-msgs ws-channel))))))
To send messages to the server, we will use the asynchronous put! function, which does not block further code execution after being called. This is a good fit for UI programming like with reagent. We will put! the new message on the send-chan channel. Separately, we will run a go-block that reads from send-chan and sends them to our websocket as they become available.
(defn send-msg
[msg]
(put! send-chan msg))
(defn send-msgs
[svr-chan]
(go-loop []
(when-let [msg (<! send-chan)]
(>! svr-chan msg)
(recur))))
(defn chat-input []
(let [v (atom nil)]
(fn []
[:div {:class "textinput"}
[:form
{:on-submit (fn [x]
(.preventDefault x)
(when-let [msg @v] (send-msg {:msg msg
:user (:user @app-state)
:m-type :chat})))}
[:input {:type "text"
:value @v
:placeholder "Type a message to send to the chatroom"
:on-change #(reset! v (-> % .-target .-value))}]
[:br]
[:button {:type "submit"} "Send"]]])))
For our client now for our client to actually receive code? Fortunately this is the easiest part of the entire application. We simply read from our websocket in a go-block and append messages to the global application state. Reagent takes care of re-rendering our view when the messages change.
(defn receive-msgs
[svr-chan]
(go-loop []
(if-let [msg (<! svr-chan)]
(let [new-msg (:message msg)]
(do
(case (:m-type new-msg)
:init-users (reset! users-set (:msg new-msg))
:chat (swap! msg-list #(conj %1 (dissoc %2 :m-type)) new-msg)
:new-user (swap! users-set #(conj %1 (:msg %2)) new-msg)
:user-left (swap! users-set #(disj %1 (:msg %2)) new-msg))
(recur)))
(println "Websocket closed"))))
Hopefully this has been a useful guided tour to some of the concepts behind core.async.
You can checkout the full code here https://github.com/briangorman/hablamos. In addition, the project has been deployed to https://chat.briangorman.io so you can see it live.
Core.async contains many other ideas and functionality and this tutorial only shows a small part of core.async. Notably, I omitted the >!! and <!! operators, which work outside of go-blocks, at the cost of blocking a normal system thread. Another major area to explore is constructing asynchronous data pipelines with the pipeline function.