gistfile1.clj

(ns benchmark-chat-server
  (:use [clojure.test])
  (:import
     [java.net InetSocketAddress]
     [java.util.concurrent Executors]
     [org.jboss.netty.bootstrap ServerBootstrap]
     [org.jboss.netty.channel Channels ChannelPipelineFactory SimpleChannelHandler]
     [org.jboss.netty.channel.group DefaultChannelGroup]
     [org.jboss.netty.channel.socket.nio NioServerSocketChannelFactory]
     [org.jboss.netty.buffer ChannelBuffers]
     [org.jboss.netty.handler.codec.frame DelimiterBasedFrameDecoder Delimiters]))

;helpers
(defn encode
  "Creates a buffer prepared for a Flash XMLSocket"
  [str]
  (ChannelBuffers/wrappedBuffer 
    (into-array 
      [(.getBytes str "UTF-8")
       (byte-array [(byte 0)])])))

;global data structures
(def users 
  {:by-name (ref {})})

(def channels 
  {:by-name (ref {})})

(def sessions '())
  
;data structures - constructors
(defn create-channel
  [channel-name]
  (dosync
    (let 
      [new-channel 
       {:channel-name channel-name
        :users (ref '())}]
      (alter (channels :by-name) conj {channel-name new-channel})
      new-channel)))

(defn create-user
  [user-name]
  (dosync
    (let 
      [new-user 
       {:user-name user-name
        :channels (ref '())
        :session (ref nil)}]
      (alter (users :by-name) conj {user-name new-user})
      new-user)))

;data structures - accessors
(defn channel-name
  [channel]
  (channel :channel-name))

(defn user-name
  [user]
  (user :user-name))

;datastrucutres - mutators
(defn add-user
  [channel user]
  (dosync
    (alter (channel :users) conj user)
    (alter (user :channels) conj channel)))

(defn user-send
  [user msg]
  (dosync
    (if @(user :session)
      (.write @(user :session) msg))))

(defn channel-join
  [channel user]
  (let
    [msg (encode (str "join " (channel-name channel) " " (user-name user)))]
    (doseq 
      [user @(channel :users)]
      (user-send user msg))))


(defn user-login
  [user netty-channel]
  (dosync
    (ref-set (user :session) netty-channel)
    (doseq [channel @(user :channels)] (channel-join channel user))))


;Netty server
(declare create-handler)

(defn start
  "Start a Netty server. Returns the server."
  [port]
  (let [all-channels (DefaultChannelGroup.
                       "server-channels")
        channel-factory (NioServerSocketChannelFactory.
                          (Executors/newCachedThreadPool)
                          (Executors/newCachedThreadPool))
        bootstrap (ServerBootstrap. channel-factory)
        pipeline (.getPipeline bootstrap)]
    (.addLast pipeline "decoder" (DelimiterBasedFrameDecoder. 8192 true (. Delimiters nulDelimiter)))
    (.addLast pipeline "handler" (create-handler all-channels))
    (.setOption bootstrap "child.tcpNoDelay", true)
    (.setOption bootstrap "child.keepAlive", true)
    (.add all-channels (.bind bootstrap (InetSocketAddress. port)))
    {:all-channels all-channels :channel-factory channel-factory}))

(defn stop
  "Stops a server"
  [server]
  (.close (server :all-channels))
  (.releaseExternalResources (server :channel-factory))
  server)

(def flash-policy-file 
  (str "<?xml version=\"1.0\"?>"
    "<!DOCTYPE cross-domain-policy SYSTEM \"http://www.adobe.com/xml/dtds/cross-domain-policy.dtd\">"
    "<cross-domain-policy>"   
    "<allow-access-from domain=\"195.198.115.136\" to-ports=\"4711\"/>"
    "</cross-domain-policy>"))

(defn login
  [args netty-channel]
  (let
    [user-name (first args)
     user (@(users :by-name) user-name)]
    (if user 
      (user-login user netty-channel)
      (str "error no such user " user-name))))

(defn handle-message
  [msg netty-channel]
  (let 
    [msg-seq (.split #" " msg)
    command (first msg-seq)
    args (rest msg-seq)]
    (cond
      (= "login" command) (login args netty-channel)
      :else (str "error Unknown Command: " command))))
  
(defn create-handler
  "Returns a Netty handler."
  [all-channels]
  (proxy [SimpleChannelHandler] []
    (channelConnected [ctx e]
      (let [c (.getChannel e)]
        (.add all-channels c)
        (println "Connected:" c)))

    (channelDisconnected [ctx e]
      (let [c (.getChannel e)]
        (println "Disconnected:" c)))

    (messageReceived [ctx e]
      (let 
        [c (.getChannel e)
         cb (.getMessage e)
         msg (.toString cb "UTF-8")]
        (let 
          [reply (cond 
                   (.isEmpty msg) ("error Empty Message")
                   (= '<' (get msg 0)) flash-policy-file
                   :else (handle-message msg c))]
          (when reply
            (do
              (println reply)
              (.write c (encode reply)))))))

    (exceptionCaught
      [ctx e]
      (let [throwable (.getCause e)]
        (.printStackTrace throwable)
        (println "@exceptionCaught " throwable))
      (-> e .getChannel .close))))


; Test code
(defn clean-fixture 
  [f]
  (dosync
    (ref-set (users :by-name) {})
    (ref-set (channels :by-name) {}))
  (f)
  (dosync
    (ref-set (users :by-name) {})
    (ref-set (channels :by-name) {}))
 )
(deftest test-create-user
  (dosync      
    (ref-set (users :by-name) {})
    (is (= 0 (count @(users :by-name))))
    (let
      [jonas (create-user "Jonas")]
      (is (= "Jonas" (user-name jonas))))
    (is (= 1 (count @(users :by-name))))))

(deftest test-create-user-2
  (dosync      
    (ref-set (users :by-name) {})
    (is (= 0 (count @(users :by-name))))
    (let
      [jonas (create-user "Jonas")
       marcus (create-user "Marcus")]
      (is (= "Jonas" (user-name jonas)))
      (is (= "Marcus" (user-name marcus))))
    (is (= 2 (count @(users :by-name))))))

(deftest test-add-user
  (let
    [jonas (create-user "Jonas")
     lobby (create-channel "Lobby")]
    (is (= 0 (count @(jonas :channels))))
    (is (= 0 (count @(lobby :users))))
    (add-user lobby jonas)
    (is (= 1 (count @(jonas :channels))))
    (is (= 1 (count @(lobby :users))))))

(run-tests 'benchmark-chat-server)

(def jonas (create-user "Jonas"))
(def marcus (create-user "Marcus"))

(def lobby (create-channel "Lobby"))

(add-user lobby jonas)
(add-user lobby marcus)