project.clj

(defproject tcp-test "0.1.0-SNAPSHOT"
  :dependencies [[org.clojure/clojure "1.4.0"]
                 [aleph "0.3.0-SNAPSHOT"]
                 [org.clojure/tools.cli "0.2.1"]]
  :main tcp-test.core)

core.clj

(ns tcp-test.core
  (:gen-class)
  (:require [clojure.tools.cli :refer :all]
            [aleph.tcp :refer :all]
            [lamina.core :refer :all]
            [gloss.core :refer :all]
            [lamina.stats :refer :all]))

(def msg (.getBytes "0123456789"))

(defn ^:private handler [channel client-info]
  (let [rate-channel (rate channel)]
    (receive-all rate-channel #(println % "fps")))
  (receive-all channel (fn [buffer]
                         (let [buf (gloss.io/contiguous buffer)
                               ary (byte-array (.remaining buf))]
                           (.get buf ary)))))

(defn start-server [& port]
  (start-tcp-server handler
                    {:port (or port 5555),
                     :frame (finite-block :int32)}))

(defn start-client [& port]
  (wait-for-result
   (tcp-client {:host "localhost",
                :port (or port 5555),
                :frame (finite-block :int32)})))

(defn run-test
  []
  (let [client-channel (start-client)
        req-batch-size 100000]
    (on-error client-channel #(println %))
    (on-closed client-channel #(println "channel closed"))
    (future
      (while true
        ;; (Thread/sleep 1000)
        (dotimes [n req-batch-size]
          (if (closed? client-channel)
            (throw (Exception. "channel_closed"))
            (enqueue client-channel msg)))))))

(defn -main
  [& args]
  (let [[options args banner]
        (cli args
             ["--server" :default false :flag true]
             ["--run-test" :default false :flag true])]
    (when (:help options)
      (println banner)
      (System/exit 0))

    (when (:server options)
      (start-server))

    (when (:run-test options)
      (run-test))))