clojerl

;   Copyright (c) Rich Hickey. All rights reserved.
;   The use and distribution terms for this software are covered by the
;   Eclipse Public License 1.0 (http://opensource.org/licenses/eclipse-1.0.php)
;   which can be found in the file epl-v10.html at the root of this distribution.
;   By using this software in any fashion, you are agreeing to be bound by
;   the terms of this license.
;   You must not remove this notice, or any other, from this software.

(ns ^{:doc "Socket server support"
      :author "Alex Miller"}
  clojure.core.server
  (:require [clojure.string :as str]
            [clojure.edn :as edn]
            [clojure.main :as m])
  (:import [clojerl String]
           [erlang.io PushbackReader]))

(set! *warn-on-infer* true)

(def ^:dynamic *session* nil)

(defonce ^:private servers (atom {}))

(defmacro process [& body]
  `(erlang/spawn (fn* [] ~@body)))

(defn- required
  "Throw if opts does not contain prop."
  [opts prop]
  (when (nil? (get opts prop))
    (throw (ex-info (str "Missing required socket server property " prop) opts))))

(defn- validate-opts
  "Validate server config options"
  [{:keys [name port accept] :as opts}]
  (doseq [prop [:name :port :accept]] (required opts prop))
  (when (or (not (integer? port)) (not (< -1 port 65535)))
    (throw (ex-info (str "Invalid socket server port: " port) opts))))

(defn- accept-connection
  "Start accept function, to be invoked on a client thread, given:
    conn - client socket
    name - server name
    client-id - client identifier
    in - in stream
    out - out stream
    err - err stream
    accept - accept fn symbol to invoke
    args - to pass to accept-fn"
  [conn name client-id in out err accept args]
  (try
    (binding [*in* in
              *out* out
              *err* err
              *session* {:server name :client client-id}]
      (swap! servers assoc-in [name :sessions client-id] {})
      (require (symbol (namespace accept)))
      (let [accept-fn (resolve accept)]
        (apply accept-fn args)))
    (catch _ disconnect)
    (finally
      (swap! servers update-in [name :sessions] dissoc client-id)
      (gen_tcp/close conn))))

(defn* unwrap-result
  ([#erl[:ok res]] res)
  ([#erl[:error error]] (throw error))
  ([res] res))

(defn* getaddr
  ([nil]
   (getaddr "localhost"))
  ([address]
   (-> address
       erlang/binary_to_list
       (inet/getaddr :inet)
       unwrap-result)))

(defn listen [address port]
  (let [opts #erl(:binary #erl[:ip address] #erl[:active false])]
    (unwrap-result (gen_tcp/listen port opts))))

(defn accept-socket [socket]
  (unwrap-result (gen_tcp/accept socket)))

(deftype SocketWriter [conn]
  erlang.io.IWriter
  (write [this data]
    (unwrap-result (gen_tcp/send conn data)))
  (write [this format values]
    (let [data (->> values (io_lib/format format) (erlang/iolist_to_binary.1))]
      (unwrap-result (gen_tcp/send conn data)))))

(deftype SocketReader [conn]
  erlang.io.IReader
  (read [this]
    (unwrap-result (gen_tcp/recv conn 1)))
  (read [this length]
    (unwrap-result (gen_tcp/recv conn length :infinity)))
  (read_line [this]
    (loop [line ""]
      (let [char (unwrap-result (gen_tcp/recv conn 1))]
        (if (= char "\n")
          (str line "\n")
          (recur (str line char))))))
  (skip [this length]
    (let [skipped (unwrap-result (gen_tcp/recv conn length))]
      (count skipped))))

(defn start-server
  "Start a socket server given the specified opts:
    :address Host or address, string, defaults to loopback address
    :port Port, integer, required
    :name Name, required
    :accept Namespaced symbol of the accept function to invoke, required
    :args Vector of args to pass to accept function
    :bind-err Bind *err* to socket out stream?, defaults to true
    :server-daemon Is server thread a daemon?, defaults to true
    :client-daemon Are client threads daemons?, defaults to true
   Returns server socket."
  [opts]
  (validate-opts opts)
  (let [{:keys [address port name accept args bind-err server-daemon client-daemon]
         :or {bind-err true
              server-daemon true
              client-daemon true}} opts
        address (getaddr address) ;; nil returns loopback
        socket  (listen address port)]
    (swap! servers assoc name {:name name, :socket socket, :sessions {}})
    (process
      (try
        (loop [client-counter 1]
          (when (erlang/port_info socket)
            (try
              (let [conn (accept-socket socket)
                    in (new PushbackReader (new SocketReader conn))
                    out (new SocketWriter conn)
                    client-id (str client-counter)]
                (process
                  (accept-connection conn name client-id in out (if bind-err out *err*) accept args)))
              (catch _ disconnect))
            (recur (inc client-counter))))
        (finally
          (swap! servers dissoc name))))
    socket))

(defn stop-server
  "Stop server with name or use the server-name from *session* if none supplied.
  Returns true if server stopped successfully, nil if not found, or throws if
  there is an error closing the socket."
  ([]
   (stop-server (:server *session*)))
  ([name]
   (let [server-socket (get-in @servers [name :socket])]
     (when server-socket
       (swap! servers dissoc name)
       (gen_tcp/close server-socket)
       true))))

(defn stop-servers
  "Stop all servers ignores all errors, and returns nil."
  []
  (doseq [name (keys @servers)]
    (future (stop-server name))))

(defn- parse-props
  "Parse clojure.server.* from properties to produce a map of server configs."
  [props]
  (reduce
    (fn [acc [^String k ^String v]]
      (let [[k1 k2 k3] (str/split k #"\.")]
        (if (and (= k1 "clojure") (= k2 "server"))
          (conj acc (merge {:name k3} (edn/read-string v)))
          acc)))
    [] props))

(defn start-servers
  "Start all servers specified in the system properties."
  [system-props]
  (doseq [server (parse-props system-props)]
    (start-server server)))

(defn repl-init
  "Initialize repl in user namespace and make standard repl requires."
  []
  (in-ns 'clje.user)
  (apply require clojure.main/repl-requires))

(defn repl-read
  "Enhanced :read hook for repl supporting :repl/quit."
  [request-prompt request-exit]
  (or ({:line-start request-prompt :stream-end request-exit}
        (m/skip-whitespace *in*))
      (let [input (read {:read-cond :allow} *in*)]
        (m/skip-if-eol *in*)
        (case input
          :repl/quit request-exit
          input))))

(defn repl
  "REPL with predefined hooks for attachable socket server."
  []
  (m/repl
    :init repl-init
    :read repl-read))