gistfile1.clj

(ns chat.server
  (:use 
    [clojure.contrib.def]
    [chat.common])
  (:require
    [chat.session])
  (: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]))
  
(defn receive
  [server ctx]
  (let 
    [netty-channel (.getChannel ctx)
     cb (.getMessage ctx)
     msg (.toString cb "UTF-8")
     session (@(server :sessions) netty-channel)]
    (if session
      (session :receive msg)
      (do
        (println "Received: " msg)
        (throw 
          (IllegalStateException. 
            (str
              "Receive on unregistered netty-channel: " 
              netty-channel)))))))

(defn- netty-exception
  [server netty-exception netty-channel]
  (let
    [exception-message (str netty-exception)]    
;    (when 
;		  (not 
;		    (= 
;		      exception-message
;		      "java.io.IOException: An existing connection was forcibly closed by the remote host"))
    (println "Exception: {" exception-message "}")))
    

(defn- connect
  [server-state netty-channel]
  (.add (server-state :all-channels) netty-channel)
  (swap! 
    (server-state :sessions) 
    conj 
    {netty-channel (chat.session/create netty-channel)}))

(defn- disconnect
  [server-state netty-channel]
  (let
    [session (@(server-state :sessions) netty-channel)]
    ;(println "Disconnect")
    (session :disconnect)
	  (swap! (server-state :sessions) dissoc netty-channel)))

(defn dispatcher
  [state message & args]
  (cond 
    (= message :state) state
    (= message :receive) (apply receive state args)
    (= message :netty-exception) (apply netty-exception state args)
    (= message :connect) (apply connect state args)
    (= message :disconnect) (apply disconnect state args)
    :else (throw (IllegalArgumentException. 
                   (str "unknown message in [" "server" "] " message)))))

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

(defn create-handler
  "Returns a Netty handler."
  [server]
  (proxy [SimpleChannelHandler] []
    (channelConnected [ctx e]
      (let [c (.getChannel e)]
        (server :connect c)))

    (channelDisconnected [ctx e]
      (let [c (.getChannel e)]
        (server :disconnect c)))

    (messageReceived [ctx e]
      (server :receive e))

    (exceptionCaught
      [ctx e]
      (let 
        [throwable (.getCause e)
         c (.getChannel e)]
        (server :netty-exception throwable c)
;        (server :send :disconnect c)
        ))))

(defn create
  [port]
  (let [all-channels (DefaultChannelGroup. "server-channels")
        channel-factory (NioServerSocketChannelFactory.
                          (Executors/newCachedThreadPool)
                          (Executors/newCachedThreadPool))
        bootstrap (ServerBootstrap. channel-factory)
        pipeline (.getPipeline bootstrap)
        new-server
        (partial dispatcher     
            {
             :sessions (atom {})
             :all-channels all-channels
             :channel-factory channel-factory
             })]
    (.addLast pipeline "decoder" (DelimiterBasedFrameDecoder. 8192 true (. Delimiters nulDelimiter)))
;    (.addLast pipeline "decoder" (DelimiterBasedFrameDecoder. 8192 true (. Delimiters lineDelimiter)))
    (.addLast pipeline "handler" (create-handler new-server))
    (.setOption bootstrap "child.tcpNoDelay", true)
    (.setOption bootstrap "child.keepAlive", true)
    (.add all-channels (.bind bootstrap (InetSocketAddress. port)))
    new-server))