Обработка клиентских сообщений websocket с помощью aleph

Question

Во время моего стремления изучить Clojure я в настоящее время сталкиваюсь с проблемами при настройке взаимодействия через websocket. После многих разных подходов я в конечном итоге использовал aleph.

Чего мне удалось добиться:

• обработка подключения нового клиента
• обработка отключения клиента
• говорить с сервера клиентам по желанию

Мне не хватает средств для запуска функции-обработчика всякий раз, когда один из подключенных клиентов отправляет что-либо через веб-сокет.

Мой код до сих пор:

(ns wonders7.core.handler
  (:require [compojure.core :refer :all]
            [compojure.route :as route]
            [ring.middleware.defaults :refer [wrap-defaults site-defaults]]
            [aleph.http :as http]
            [manifold.stream :as stream]
            [clojure.tools.logging :refer [info]]))

(defn uuid [] (str (java.util.UUID/randomUUID)))

(def clients (atom {}))

(defn ws-create-handler [req]
  (let [ws @(http/websocket-connection req)]
    (info "ws-create-handler")
    (stream/on-closed ws #(swap! clients dissoc ws))
    (swap! clients assoc ws (uuid))))

(defroutes app-routes
  (GET "/ws" [] ws-create-handler)
  (route/not-found "Not Found"))

(def app
  (wrap-defaults app-routes site-defaults))

(defn msg-to-client [[client-stream uuid]]
  (stream/put! client-stream "The server side says hello!"))

(defn msg-broadcast []
  (map #(msg-to-client %) @clients))

;(stream/take! (first (first @clients)))
;(http/start-server app {:port 8080})

Я запускаю сервер Netty с закомментированным вызовом aleph http / start-server. Мне также удалось получить сообщения от клиента через ручную стрим / дубль! вызов (также закомментирован). То, что мне нужно выяснить, это как автоматически запустить этот процесс, когда что-то приходит.

Заранее благодарю за любую помощь!

Answer 1

Функция, которую вы ищете (manifold.stream/consume callback stream), который будет вызывать обратный вызов для каждого сообщения, которое выходит из потока.

Answer 2

В этом примере автор использует recieve-all а также siphon от aleph, чтобы выполнить очень похожую задачу, которую я примерно перефразирую как:

(let [chat (named-channel room (receive-all ch #(println "message: " %)))]
  (siphon chat ch)