message_dispatcher.clj

(ns de.mpg.shh.util-message-server.message-dispatcher
  (:require [clojure.tools.logging :refer [info error]]
            [clojure.string :as str]
            [clojure.core.async :as async]
            [com.stuartsierra.component :as component]
            [de.mpg.shh.util-message-server.life-cycle :as life-cycle])
  (:import [java.io StringWriter PrintWriter]))

(defn stack-trace-to-string
  "Returns a string containing the output of .printStackTrace"
  [t]
  (let [sw (StringWriter.)
        pw (PrintWriter. sw)
        _ (.printStackTrace t pw)]
    (.toString sw)))

(defn report-error-truthfully
  [request-channel end-points error-msg]
  (if (contains? end-points :error)
    (try
      (do
        ((get end-points :error) request-channel error-msg)
        true)
      (catch Throwable t
        (error (stack-trace-to-string t))
        true))
    (do
      (error error-msg)
      true)))

;; This provides a convenience dispatcher, just supply a map of end points
(defn dispatch [request-channel end-points ?data]
  (if (vector? ?data)
    (if (= (first ?data) :stop)
      false
      (if (contains? end-points (first ?data))
        (try
          (do
            ((get end-points (first ?data)) request-channel (second ?data))
            true)
          (catch Throwable t
            (report-error-truthfully request-channel end-points (stack-trace-to-string t))))
        (report-error-truthfully request-channel end-points (str/join "" ["Unhandled event " (first ?data) " not found in end-points map"]))))
    (report-error-truthfully request-channel end-points (str/join "" ["Event format not recognised " (str ?data)]))))

(defrecord MessageDispatcher [listen-context-config listen-address listen-selector dispatcher-request-buffer-size stop-message]
  component/Lifecycle
  (start [{{request-channel :request-channel
            :as message-server} :message-server
           dispatcher-request-transducer :dispatcher-request-transducer
           end-points :end-points
           :as component}]
         (let [dispatcher-request-channel (async/chan dispatcher-request-buffer-size dispatcher-request-transducer)
               _ (life-cycle/listen message-server listen-context-config listen-address dispatcher-request-channel listen-selector)]
    (async/go-loop [dispatcher-request (async/<! dispatcher-request-channel)]
                   (if (dispatch request-channel end-points dispatcher-request)
                     (recur (async/<! dispatcher-request-channel))
                     (info "Shutdown dispatcher request channel")))
           (assoc component :dispatcher-request-channel dispatcher-request-channel
                            :stop-message stop-message
                            :dispatcher-request-buffer-size dispatcher-request-buffer-size)))
  (stop [{dispatcher-request-channel :dispatcher-request-channel
          stop-message :stop-message
          :as component}]
        (async/>!! dispatcher-request-channel stop-message)))