Server

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
use "collections"
use "net"
use "net_ssl"
use "time"
use "debug"

interface tag _SessionRegistry
  be register_session(conn: _ServerConnection)
  be unregister_session(conn: _ServerConnection)

actor Server is _SessionRegistry
  """
  Runs an HTTP server.

  ### Server operation

  Information flow into the Server is as follows:

  1. `Server` listens for incoming TCP connections.

  2. `_ServerConnHandler` is the notification class for new connections. It creates
  a `_ServerConnection` actor and receives all the raw data from TCP. It uses
  the `HTTP11RequestParser` to assemble complete `Request` objects which are passed off
  to the `_ServerConnection`.

  3. The `_ServerConnection` actor deals with requests and their bodies
  that have been parsed by the `HTTP11RequestParser`. This is where requests get
  dispatched to the caller-provided Handler.
  """
  let _notify: ServerNotify
  var _handler_maker: HandlerFactory val
  let _config: ServerConfig
  let _sslctx: (SSLContext | None)
  let _listen: TCPListener
  var _address: NetAddress
  let _sessions: SetIs[_ServerConnection tag] = SetIs[_ServerConnection tag]
  let _timers: Timers = Timers
  var _timer: (Timer tag | None) = None

  new create(
    auth: TCPListenAuth,
    notify: ServerNotify iso,
    handler: HandlerFactory val,
    config: ServerConfig,
    sslctx: (SSLContext | None) = None)
  =>
    """
    Create a server bound to the given host and service. To do this we
    listen for incoming TCP connections, with a notification handler
    that will create a server session actor for each one.
    """
    _notify = consume notify
    _handler_maker = handler
    _config = config
    _sslctx = sslctx
    Debug("starting server with config:\n" + config.to_json())

    _listen = TCPListener(auth,
        _ServerListener(this, config, sslctx, _handler_maker),
        config.host, config.port, config.max_concurrent_connections)

    _address = recover NetAddress end

  be register_session(conn: _ServerConnection) =>
    _sessions.set(conn)

    // only start a timer if we have a connection-timeout configured
    if _config.has_timeout() then
      match _timer
      | None =>
        let that: Server tag = this
        let timeout_interval = _config.timeout_heartbeat_interval
        let t = Timer(
          object iso is TimerNotify
            fun ref apply(timer': Timer, count: U64): Bool =>
              that._start_heartbeat()
              true
          end,
          Nanos.from_millis(timeout_interval),
          Nanos.from_millis(timeout_interval))
        _timer = t
        _timers(consume t)
      end
    end

  be _start_heartbeat() =>
    // iterate through _sessions and ping all connections
    let current_seconds = Time.seconds() // seconds resolution is fine
    for session in _sessions.values() do
      session._heartbeat(current_seconds)
    end

  be unregister_session(conn: _ServerConnection) =>
    _sessions.unset(conn)

  be set_handler(handler: HandlerFactory val) =>
    """
    Replace the request handler.
    """
    _handler_maker = handler
    _listen.set_notify(
      _ServerListener(this, _config, _sslctx, _handler_maker))

  be dispose() =>
    """
    Shut down the server gracefully. To do this we have to eliminate
    any source of further inputs. So we stop listening for new incoming
    TCP connections, and close any that still exist.
    """
    _listen.dispose()
    _timers.dispose()
    for conn in _sessions.values() do
      conn.dispose()
    end

  fun local_address(): NetAddress =>
    """
    Returns the locally bound address.
    """
    _address

  be _listening(address: NetAddress) =>
    """
    Called when we are listening.
    """
    _address = address
    _notify.listening(this)

  be _not_listening() =>
    """
    Called when we fail to listen.
    """
    _notify.not_listening(this)

  be _closed() =>
    """
    Called when we stop listening.
    """
    _notify.closed(this)