stdlib/gserver.rb

   1 #
   2 # Copyright (C) 2001 John W. Small All Rights Reserved
   3 #
   4 # Author::        John W. Small
   5 # Documentation:: Gavin Sinclair
   6 # Licence::       Freeware.
   7 #
   8 # See the class GServer for documentation.
   9 #
  10
  11 require "socket"
  12 require "thread"
  13
  14 #
  15 # GServer implements a generic server, featuring thread pool management,
  16 # simple logging, and multi-server management.  See HttpServer in
  17 # <tt>xmlrpc/httpserver.rb</tt> in the Ruby standard library for an example of
  18 # GServer in action.
  19 #
  20 # Any kind of application-level server can be implemented using this class.
  21 # It accepts multiple simultaneous connections from clients, up to an optional
  22 # maximum number.  Several _services_ (i.e. one service per TCP port) can be
  23 # run simultaneously, and stopped at any time through the class method
  24 # <tt>GServer.stop(port)</tt>.  All the threading issues are handled, saving
  25 # you the effort.  All events are optionally logged, but you can provide your
  26 # own event handlers if you wish.
  27 #
  28 # === Example
  29 #
  30 # Using GServer is simple.  Below we implement a simple time server, run it,
  31 # query it, and shut it down.  Try this code in +irb+:
  32 #
  33 #   require 'gserver'
  34 #
  35 #   #
  36 #   # A server that returns the time in seconds since 1970.
  37 #   #
  38 #   class TimeServer < GServer
  39 #     def initialize(port=10001, *args)
  40 #       super(port, *args)
  41 #     end
  42 #     def serve(io)
  43 #       io.puts(Time.now.to_i)
  44 #     end
  45 #   end
  46 #
  47 #   # Run the server with logging enabled (it's a separate thread).
  48 #   server = TimeServer.new
  49 #   server.audit = true                  # Turn logging on.
  50 #   server.start
  51 #
  52 #   # *** Now point your browser to http://localhost:10001 to see it working ***
  53 #
  54 #   # See if it's still running.
  55 #   GServer.in_service?(10001)           # -> true
  56 #   server.stopped?                      # -> false
  57 #
  58 #   # Shut the server down gracefully.
  59 #   server.shutdown
  60 #
  61 #   # Alternatively, stop it immediately.
  62 #   GServer.stop(10001)
  63 #   # or, of course, "server.stop".
  64 #
  65 # All the business of accepting connections and exception handling is taken
  66 # care of.  All we have to do is implement the method that actually serves the
  67 # client.
  68 #
  69 # === Advanced
  70 #
  71 # As the example above shows, the way to use GServer is to subclass it to
  72 # create a specific server, overriding the +serve+ method.  You can override
  73 # other methods as well if you wish, perhaps to collect statistics, or emit
  74 # more detailed logging.
  75 #
  76 #   connecting
  77 #   disconnecting
  78 #   starting
  79 #   stopping
  80 #
  81 # The above methods are only called if auditing is enabled.
  82 #
  83 # You can also override +log+ and +error+ if, for example, you wish to use a
  84 # more sophisticated logging system.
  85 #
  86 class GServer
  87
  88   DEFAULT_HOST = "127.0.0.1"
  89
  90   def serve(io)
  91   end
  92
  93   @@services = {}   # Hash of opened ports, i.e. services
  94   @@servicesMutex = Mutex.new
  95
  96   def GServer.stop(port, host = DEFAULT_HOST)
  97     @@servicesMutex.synchronize {
  98       @@services[host][port].stop
  99     }
 100   end
 101
 102   def GServer.in_service?(port, host = DEFAULT_HOST)
 103     @@services.has_key?(host) and
 104       @@services[host].has_key?(port)
 105   end
 106
 107   def stop
 108     @connectionsMutex.synchronize  {
 109       if @tcpServerThread
 110         @tcpServerThread.raise "stop"
 111       end
 112     }
 113   end
 114
 115   def stopped?
 116     @tcpServerThread == nil
 117   end
 118
 119   def shutdown
 120     @shutdown = true
 121   end
 122
 123   def connections
 124     @connections.size
 125   end
 126
 127   def join
 128     @tcpServerThread.join if @tcpServerThread
 129   end
 130
 131   attr_reader :port, :host, :maxConnections
 132   attr_accessor :stdlog, :audit, :debug
 133
 134   def connecting(client)
 135     addr = client.peeraddr
 136     log("#{self.class.to_s} #{@host}:#{@port} client:#{addr[1]} " +
 137         "#{addr[2]}<#{addr[3]}> connect")
 138     true
 139   end
 140
 141   def disconnecting(clientPort)
 142     log("#{self.class.to_s} #{@host}:#{@port} " +
 143       "client:#{clientPort} disconnect")
 144   end
 145
 146   protected :connecting, :disconnecting
 147
 148   def starting()
 149     log("#{self.class.to_s} #{@host}:#{@port} start")
 150   end
 151
 152   def stopping()
 153     log("#{self.class.to_s} #{@host}:#{@port} stop")
 154   end
 155
 156   protected :starting, :stopping
 157
 158   def error(detail)
 159     log(detail.backtrace.join("\n"))
 160   end
 161
 162   def log(msg)
 163     if @stdlog
 164       @stdlog.puts("[#{Time.new.ctime}] %s" % msg)
 165       @stdlog.flush
 166     end
 167   end
 168
 169   protected :error, :log
 170
 171   def initialize(port, host = DEFAULT_HOST, maxConnections = 4,
 172     stdlog = $stderr, audit = false, debug = false)
 173     @tcpServerThread = nil
 174     @port = port
 175     @host = host
 176     @maxConnections = maxConnections
 177     @connections = []
 178     @connectionsMutex = Mutex.new
 179     @connectionsCV = ConditionVariable.new
 180     @stdlog = stdlog
 181     @audit = audit
 182     @debug = debug
 183   end
 184
 185   def start(maxConnections = -1)
 186     raise "running" if !stopped?
 187     @shutdown = false
 188     @maxConnections = maxConnections if maxConnections > 0
 189     @@servicesMutex.synchronize  {
 190       if GServer.in_service?(@port,@host)
 191         raise "Port already in use: #{host}:#{@port}!"
 192       end
 193       @tcpServer = TCPServer.new(@host,@port)
 194       @port = @tcpServer.addr[1]
 195       @@services[@host] = {} unless @@services.has_key?(@host)
 196       @@services[@host][@port] = self;
 197     }
 198     @tcpServerThread = Thread.new {
 199       begin
 200         starting if @audit
 201         while !@shutdown
 202           @connectionsMutex.synchronize  {
 203              while @connections.size >= @maxConnections
 204                @connectionsCV.wait(@connectionsMutex)
 205              end
 206           }
 207           client = @tcpServer.accept
 208           @connections << Thread.new(client)  { |myClient|
 209             begin
 210               myPort = myClient.peeraddr[1]
 211               serve(myClient) if !@audit or connecting(myClient)
 212             rescue => detail
 213               error(detail) if @debug
 214             ensure
 215               begin
 216                 myClient.close
 217               rescue
 218               end
 219               @connectionsMutex.synchronize {
 220                 @connections.delete(Thread.current)
 221                 @connectionsCV.signal
 222               }
 223               disconnecting(myPort) if @audit
 224             end
 225           }
 226         end
 227       rescue => detail
 228         error(detail) if @debug
 229       ensure
 230         begin
 231           @tcpServer.close
 232         rescue
 233         end
 234         if @shutdown
 235           @connectionsMutex.synchronize  {
 236              while @connections.size > 0
 237                @connectionsCV.wait(@connectionsMutex)
 238              end
 239           }
 240         else
 241           @connections.each { |c| c.raise "stop" }
 242         end
 243         @tcpServerThread = nil
 244         @@servicesMutex.synchronize  {
 245           @@services[@host].delete(@port)
 246         }
 247         stopping if @audit
 248       end
 249     }
 250     self
 251   end
 252
 253 end