Pass channel socket to server message handler

[hugoduncan]

When using Sente with a component model, the channel socket's send function is not globally available. To make the send-fn available in the message handler, would it be possible to add the channel socket to the map passed as the first argument to the message handler? This would also imply changing start-chsk-router-loop! to take the channel socket rather than ch-chsk.

This obviates the need to pass the ch-recv function to the message handler as a separate argument, which makes writing message handling middleware (à la ring, but without any repsonse handling) somewhat nicer.

This would be a breaking change, so could be introduced as another function, in parallel to the existing start-chsk-router-loop! function (which could be deprecated).

Current workarounds for this I have used are a custom start-chsk-router-loop! function implementing the above, or adding the channel socket to the ring session (which is available to the message handler).

[danielsz]

Hi Hugo,

With a component model, you can use Sente like system does, which includes the send function and everything you might need.

I hope this helps.

[hugoduncan]

Hi Daniel,

Making the channel-socket available from the component is certainaly another workaround, but still requires manually "wiring" into the message handler.

[danielsz]

Hi Hugo,

Yes, you pass the message handler to the component when you assemble a system. Is that not optimal?

[hugoduncan]

@danielsz

The sente component you reference already wires the message handler to the sente dispatch loop in it's start function. Having to do separate wiring outside of the component seems like extra work.

If I'm reading your component correctly, the channel socket are not assoc'd onto the component until the component is started. I'm not sure how you could use that during the system assembly in a straightforward manner.

Seems sub-optimal to me.

[danielsz]

Here's a usage example:

(defn dev-system []
  (component/system-map
   :sente (new-channel-sockets event-msg-handler)
   :web (new-web-server (Integer. (env :http-port)) app)
   :message-broker (new-rabbit-mq (env :rabbit-mq-uri))))

Where event-msg-handler is defined in a separate namespace, usually along with the app Var.

(defn event-msg-handler
  [{:as ev-msg :keys [ring-req event ?reply-fn]} _] ... ))

What am I missing?

[hugoduncan]

In event-msg-handler I would like to start an async loop that pushes messages to the client. To do that requires access to the :send-fn key in the channel socket, iiuc. I don't see how that can happen as things are now, but maybe I'm missing something obvious.

[danielsz]

You can access the :send-fn function in event-msg-handler or anywhere else in your application by referring to it through the system map.

(ns my.namespace
  (:require [reloaded.repl :refer [system]]))

(defn event-msg-handler
  [{:as ev-msg :keys [ring-req event ?reply-fn]} _] 
  (let [session (:session ring-req)
        uid     (:uid session)
        [id data :as ev] event]
    (match [id data]
        [:myapp/api {:action :users}] 
            ((:chsk-send! (:sente system)) (:uid (:session request)) [:myapp/api {:action "rejoice"}])
           :else
           (do ;(println "Unmatched event: %s" ev)
               (when-not (:dummy-reply-fn? (meta ?reply-fn))
                 (?reply-fn {:umatched-event-as-echoed-from-from-server ev})))))))

[hugoduncan]

event-msg-handler is a function which needs to be passed in to construct the system, so to access :chsk-send from the started system map would involve event-msg-handler closing over an atom (or similar) that is reset! with the system value once the system has been started, iiuc.

[danielsz]

No, the function doesn't get evaluated when you construct the component, but rather when there is a match in the event loop or when the buffer is read in the go block. I've updated my little pseudo-code snippet above with the scenario of access in the event-msg-handler. If you want, you're welcome to submit pseudo-code for your scenario, and we'll step through it together.

[hugoduncan]

Sorry, I seem not to be explaining myself very well. How is system constructed in your example (given that it needs to be passed the event-msg-handler)? How is it not a global var?

My component constructor looks something like this:

(defn server [options]
  (let [sente (sente-component {:msg-handler #'event-msg-handler})]
    ;; at this point the sente-component hasn't been started, so doesn't contain
    ;; the :chsk-send! key, so how does it get passed into event-msg-handler?
    (my-app-component ; or the Component system constructor
     {:sente sente})))

(start (server {})) ; 

[danielsz]

Not a problem. :-)

First, system being a library, you reference it in your project.clj, like any other library.

  :dependencies [[org.danielsz/system "0.1.0-SNAPSHOT"]
                           [http-kit "2.1.19-SNAPSHOT"]
                           [com.taoensso/sente "0.15.1"]
                            ...  ]

Systems are constructed like so:

(ns my-app.systems
  (:require 
   [com.stuartsierra.component :as component]
   (system.components 
    [http-kit :refer [new-web-server]]
    [sente :refer [new-channel-sockets]])
   [my-app.web :refer [handler event-msg-handler]]
   [environ.core :refer [env]]))

(defn dev-system []
  (component/system-map
   :sente (new-channel-sockets #'event-msg-handler)
   :web (new-web-server (Integer. (env :http-port)) #'handler)))

And they are started like so:

(ns user
  (:require [reloaded.repl :refer [system init start stop go reset]]
            [my-app.systems :refer [dev-system]]))

(reloaded.repl/set-init! dev-system)

You then type (go) in the repl. All components are started for you, and everything is accessible in the system map.

There is more info in the system's README, including tips for starting a system in production.

[hugoduncan]

I do understand how the component library works. I'm afraid I still don't see how the :chsk-send! key is passed to event-msg-handler in a componentised manner (without event-msg-handler refering directly to a singleton global system, in the case above, dev-system)).

[danielsz]

I'm sure I'm missing something obvious, too, but aren't components meant to be referred to from the system map once they are initialised? I think our best bet would be to discuss code with the scenario you're trying to implement, running or broken. I'd be happy to look at a gist or temporary repository and take it from there.

[hugoduncan]

The benefits of componentised code seem applicable to the event-msg-handler too, so why force it to depend on a singleton system instance?

The code I have, that works (not using component), is this:

(defsystem Server [:server :sente :channel-socket])

(defn server
  [options]
  (let [sock (atom nil)
        sente (sente/sente {:msg-handler #'msg-handler
                       :announce-fn #(reset! sock %)})]
    (map->Server
     {:channel-socket sock
      :sente sente
      :server (httpkit/httpkit (merge options {:handler (ui-app sock)}))})))

My sente component calls the value of :announce-fn when it is started, with the channel socket as argument. My ui-app function puts the channel socket onto the ring request via middleware, from where the msg-handler can read it.

What I would like to write is:

(defsystem Server [:server :sente])

(defn server
  [options]
  (let [sente (sente/sente {:msg-handler #'msg-handler})]
    (map->Server
     {:sente sente
      :server (httpkit/httpkit (merge options {:handler (ui-app)}))})))

and just have sente pass the channel socket as part of the first map argument to the message handler. This removes a fair bit of complexity (an atom, an announce function, and middleware to set the channel socket on the request).

As it is, the ch-recv key of the channel socket is treated specially, and passed as a separate argument to the message handler. I am just suggesting removing the special treatment for ch-recv and passing the whole channel socket map as part of the first map argument.

(defn msg-handler
  [{:as ev-msg :keys [ring-req event ?reply-fn channel-socket]}]
  …)

[ptaoussanis]

Hi Hugo, finally had a chance to look at this (again, apologies for the delay) - looks like a welcome change to me, thank you!

Am on Sente all of today; just working my way through some backlogged changes now - expecting to be able to cut a new release shortly.

[hugoduncan]

Peter, thanks for considering this. The pull request is a little more extensive than I originally thought it would be.

[ptaoussanis]

No worries, the change is good - I've been fiddling with the branches. Will merge manually, but leaving this open till I'm done :-)

[ptaoussanis]

Closing this, will reply to the PR shortly.

[hugoduncan]

Peter, are you against doing this on the clojurescript side as well? The same issues re components exist there too (an example workardound).

[ptaoussanis]

Hi Hugo, not at all against doing so - planning to experiment with some changes soon. Combination of short time + an unexpected refactor has made it take longer than expected, but will update you as soon as I have a release ready to go.

Thanks again for your input on this, you're welcome to keep it coming :-)

[ptaoussanis]

Okay, finally ready - sorry this took so long! Work is on the dev branch (changelog), pending release as v1.0.0.

As proposed, the relevant change is here:

(defn start-chsk-router!
  "Creates a go-loop to call `(event-msg-handler <event-msg>)` and returns a
  `(fn stop! [])`. Advanced users may choose to instead write their own loop
  against `ch-recv`."
  [ch-recv event-msg-handler]
  (let [ch-ctrl (chan)]
    (go-loop []
      (when-not
        (try ; Returns nil or ::stop
          (let [[v p] (async/alts! [ch-recv ch-ctrl])]
            (if (identical? p ch-ctrl) ::stop
              (let [event-msg v]
                (try
                  (tracef "Pre-handler event-msg: %s" event-msg)
                  (assert (event-msg? v))
                  (event-msg-handler event-msg)
                  nil
                  (catch #+clj Throwable #+cljs :default t
                    (errorf #+clj t
                      "Chsk router handling error: %s" event-msg))))))
          (catch #+clj Throwable #+cljs :default t
            (errorf #+clj t
              "Chsk router channel error!")))
        (recur)))
    (fn stop! [] (async/close! ch-ctrl))))

;;;; Deprecated

#+clj
(defn start-chsk-router-loop!
  "DEPRECATED: Please use `start-chsk-router!` instead."
  [event-msg-handler ch-recv]
  (start-chsk-router! ch-recv
    ;; Old handler form: (fn [ev-msg ch-recv])
    (fn [ev-msg] (event-msg-handler ev-msg (:ch-recv ev-msg)))))

#+cljs
(defn start-chsk-router-loop!
  "DEPRECATED: Please use `start-chsk-router!` instead."
  [event-handler ch-recv]
  (start-chsk-router! ch-recv
    ;; Old handler form: (fn [ev ch-recv])
    (fn [ev-msg] (event-handler (:event ev-msg) (:ch-recv ev-msg)))))

Interesting changes:

• start-chsk-router-loop! has been deprecated for start-chsk-router!.
• There used to be separate defs for client/server, now there's only one.
• Handlers now all operate on a single arg, and that arg is an event-msg.
• The content varies client/server:

(defn event-msg? [x]
  #+cljs
  (and
    (map? x)
    (encore/keys= x #{:ch-recv :send-fn :state :event})
    (let [{:keys [ch-recv send-fn state event]} x]
      (and
        (chan?        ch-recv)
        (ifn?         send-fn)
        (encore/atom? state)
        (event?       event))))

  #+clj
  (and
    (map? x)
    (encore/keys= x
      #{:ch-recv :push-fn :connected-uids
        :client-uuid :ring-req :event :?reply-fn})
    (let [{:keys [ch-recv push-fn connected-uids
                  client-uuid ring-req event ?reply-fn]} x]
      (and
        (chan?        ch-recv)
        (ifn?         push-fn)
        (encore/atom? connected-uids)
        ;; (string? client-uuid)
        (encore/nblank-str? client-uuid) ; Set by client (ajax) or server (ws)
        (map?   ring-req)
        (event? event)
        (or (nil? ?reply-fn) (ifn? ?reply-fn))))))

Would appreciate a confirmation if that'd help address your use case? Thanks again for the assistance, cheers! :-)