videocapture.network - Videocapture

(s/def ::socket (partial instance? Socket))
(s/def ::server-socket (partial instance? ServerSocket))

<<type-from-table(table=net-message-type-tbl,type="::net-message")>>

<<workstation-data-type>>
<<message-data-type>>

(s/def ::data (s/or :workstation ::workstation-data
		    :message ::message))

(defn handle-message [reader writer & [address]]
  (go-loop []
    (let [message (s/conform ::message (<! reader))
	  msg-format (:format message)
	  msg-type (:type message)
	  msg-data (:data message)]
      (debug "Handling message" message)
      (debug @network)
      (case msg-format
	:camera-data (case msg-type :request (do (>! writer {:format :camera-data
							     :type :response
							     :data {:workstation (settings/get-setting :form :room)
								    :camera (settings/get-setting :camera)}})
						 (debug "Sending camera data to supervisor."))
			   :response (do (swap! (:sockets @network)
						#(map (fn [x]
							(if (= address (:address x))
							  (assoc x :workstation (:workstation msg-data)
								 :camera (:camera msg-data))
							  x))
						      %))
					 (debug "Adding workstation:" msg-data "to" (pr-str @(:sockets @network)))))
	:message (case msg-type :request (if (= :supervisor (:type @network))
					   (swap! (:messages @(:active-workstation @network)) conj msg-data)
					   (swap! (:messages @network) conj msg-data)))))
    (debug "Done handling message.")
    (recur)))

(defn handle-text [text]
  (swap! (:messages @(:active-workstation @network)) conj text)
  (async/put! (:writer @(:active-workstation @network)) {:format :message :type :request :data text}))

(defn workstation-handle [active? socket {:keys [read write]}]
  {:pre [(s/valid? ::socket socket)]}
  (.setKeepAlive socket true)
  (let [reader (BufferedReader. (InputStreamReader. (.getInputStream socket)))
	writer (PrintWriter. ^OutputStream (.getOutputStream socket) true)]
    (go-loop []
      (let [str (edn/read-string (.readLine reader))]
	(when-not (nil? str)
	  (>! read str)))
      (when (.isConnected socket) (recur)))
    (go-loop []
      (.println writer ^String (pr-str (<! write)))
      (when (.isConnected socket) (recur)))
    (handle-message read write)
    (go-loop []
      (if-not @active?
	(.close socket)
	(recur)))))

(defn workstation-init []
  (let [socket (atom nil)
	active? (atom true)
	channels {:read (chan 2)
		  :write (chan 2)}
	messages (atom [])]
    (go-loop []
      (reset! socket (s/conform ::socket
				(try (Socket. ^String (settings/get-setting :supervisor :ip) ^Integer (settings/get-setting :supervisor :port))
				     (catch Exception e (trace e)))))
      (if (= :clojure.spec.alpha/invalid @socket)
	(do (Thread/sleep 5000)
	    (when @active? (recur)))
	(workstation-handle active? @socket channels)))
    {:active? active?
     :socket socket
     :channels channels
     :messages messages
     :type :workstation}))

(defn supervisor-handle "Establishes the new connection, adding channel interfaces and storing it in the mount state."
  [sockmap socket]
  {:pre [(s/valid? ::socket socket)]}
  (.setKeepAlive socket true)
  (let [reader (BufferedReader. (InputStreamReader. (.getInputStream socket)))
	writer (PrintWriter. ^OutputStream (.getOutputStream socket) true)
	reader-chan (chan 2)
	writer-chan (chan 2)
	messages (atom [])
	address (.getHostAddress ^InetAddress (.getInetAddress socket))]
    ;; Reader loop
    (go-loop []
      (let [str (edn/read-string (.readLine reader))]
	(when-not (nil? str)
	  (>! reader-chan str)))
      (when (.isConnected socket) (recur)))
    ;; Writer loop
    (go-loop []
      (.println writer ^String (pr-str (<! writer-chan)))
      (when (.isConnected socket) (recur)))
    ;; Check for disconnects
    (go-loop []
      (if-not (.isConnected socket)
	(swap! sockmap dissoc address)
	(recur)))
    ;; Add message parser
    (handle-message reader-chan writer-chan address)
    ;; Request camera data
    (debug "Requesting camera data from new workstation.")
    (async/put! writer-chan {:format :camera-data
			     :type :request
			     :data nil})
    ;; Associations
    (swap! sockmap remove #(= address (:address %)))
    (swap! sockmap conj
	   {:address address
	    :reader reader-chan
	    :writer writer-chan
	    :messages messages
	    :socket socket})))

(defn supervisor-init []
  (let [server-socket (s/conform ::server-socket (ServerSocket. (settings/get-setting :supervisor :port)))
	active-workstation (atom nil)
	active? (atom true)
	sockets (atom [])
	messages (atom [])]
    (if (= :s/invalid server-socket)
      (fatal "Unable to open server-socket.")
      (go-loop []
	(try (supervisor-handle sockets (.accept server-socket))
	     (catch SocketException e (reset! active? false)))
	(when @active? (recur))))
    {:active? active?
     :sockets sockets
     :server-socket server-socket
     :active-workstation active-workstation
     :type :supervisor}))

(ns videocapture.network
  (:require [videocapture.settings :as settings]
	    [clojure.edn :as edn]
	    [clojure.core.async :as async :refer [chan <! >! go go-loop]]
	    [clojure.spec.alpha :as s]
	    [mount.core :as mount]
	    [taoensso.timbre :as timbre
	     :refer [log trace debug info warn error fatal report
		     logf tracef debugf infof warnf errorf fatalf reportf
		     spy get-env]])
  (:import (java.net ServerSocket Socket InetAddress ConnectException SocketException)
	   (java.io BufferedReader InputStreamReader PrintWriter OutputStream)))

(declare network)

<<socket-types>>

<<message-types>>

<<net-messages>>

<<supervisor>>

<<workstation>>

(defn net-init []
  (if (settings/get-setting :supervisor :activated)
    (supervisor-init)
    (workstation-init)))

(defn net-stop [network]
  (reset! (:active @network) false)
  (if (= :supervisor (:type @network))
    (.close (:server-socket @network))
    (.close @(:socket @network))))

(mount/defstate network
   :start (net-init)
   :stop (net-stop network))