basis/io/servers/servers.factor

   1 ! Copyright (C) 2003, 2010 Slava Pestov, Doug Coleman.
   2 ! See http://factorcode.org/license.txt for BSD license.
   3 USING: accessors arrays calendar concurrency.flags
   4 concurrency.semaphores continuations destructors io io.sockets
   5 io.sockets.secure io.streams.duplex io.styles io.timeouts kernel
   6 logging make math namespaces present prettyprint random
   7 sequences sets strings threads ;
   8 IN: io.servers
   9
  10 TUPLE: threaded-server < identity-tuple
  11 name
  12 log-level
  13 secure
  14 insecure
  15 secure-config
  16 servers
  17 max-connections
  18 semaphore
  19 timeout
  20 encoding
  21 handler
  22 server-stopped
  23 secure-context ;
  24
  25 SYMBOL: running-servers
  26 running-servers [ HS{ } clone ] initialize
  27
  28 ERROR: server-not-running threaded-server ;
  29
  30 ERROR: server-already-running threaded-server ;
  31
  32 <PRIVATE
  33
  34 : add-running-server ( threaded-server running-servers -- )
  35     dupd ?adjoin [ drop ] [ server-already-running ] if ;
  36
  37 : remove-running-server ( threaded-server running-servers -- )
  38     dupd ?delete [ drop ] [ server-not-running ] if ;
  39
  40 PRIVATE>
  41
  42 : local-server ( port -- addrspec ) "localhost" swap <inet> ;
  43
  44 : internet-server ( port -- addrspec ) f swap <inet> ;
  45
  46 : new-threaded-server ( encoding class -- threaded-server )
  47     new
  48         "server" >>name
  49         DEBUG >>log-level
  50         <secure-config> >>secure-config
  51         1 minutes >>timeout
  52         [ "No handler quotation" throw ] >>handler
  53         swap >>encoding ;
  54
  55 : <threaded-server> ( encoding -- threaded-server )
  56     threaded-server new-threaded-server ;
  57
  58 GENERIC: handle-client* ( threaded-server -- )
  59
  60 <PRIVATE
  61
  62 GENERIC: >insecure ( obj -- obj )
  63
  64 M: inet >insecure 1array ;
  65 M: inet4 >insecure 1array ;
  66 M: inet6 >insecure 1array ;
  67 M: local >insecure 1array ;
  68 M: integer >insecure internet-server 1array ;
  69 M: string >insecure internet-server 1array ;
  70 M: array >insecure [ >insecure ] map concat ;
  71 M: f >insecure ;
  72
  73 : >secure ( addrspec -- addrspec' )
  74     >insecure [ dup secure? [ f <secure> ] unless ] map ;
  75
  76 : configurable-addrspecs ( addrspecs -- addrspecs' )
  77     [ inet6? not ipv6-supported? or ] filter ;
  78
  79 : listen-on ( threaded-server -- addrspecs )
  80     [ secure>> ssl-supported? [ >secure ] [ drop { } ] if ]
  81     [ insecure>> >insecure ] bi append
  82     [ resolve-host ] map concat configurable-addrspecs ;
  83
  84 : accepted-connection ( remote local -- )
  85     [
  86         [ "remote: " % present % ", " % ]
  87         [ "local: " % present % ]
  88         bi*
  89     ] "" make
  90     \ accepted-connection NOTICE log-message ;
  91
  92 : log-connection ( remote local -- )
  93     [ accepted-connection ]
  94     [ [ remote-address namespaces:set ] [ local-address namespaces:set ] bi* ]
  95     2bi ;
  96
  97 M: threaded-server handle-client* handler>> call( -- ) ;
  98
  99 : handle-client ( client remote local -- )
 100     '[
 101         _ _ log-connection
 102         threaded-server get
 103         [ timeout>> timeouts ] [ handle-client* ] bi
 104     ] with-stream ;
 105
 106 \ handle-client NOTICE add-error-logging
 107
 108 : client-thread-name ( addrspec -- string )
 109     [ threaded-server get name>> ] dip
 110     unparse-short " connection from " glue ;
 111
 112 : (accept-connection) ( server -- )
 113     [ accept ] [ addr>> ] bi
 114     [ '[ _ _ _ handle-client ] ]
 115     [ drop client-thread-name ] 2bi
 116     spawn drop ;
 117
 118 : accept-connection ( server -- )
 119     threaded-server get semaphore>>
 120     [ [ (accept-connection) ] with-semaphore ]
 121     [ (accept-connection) ]
 122     if* ;
 123
 124 : with-existing-secure-context ( threaded-server quot -- )
 125     [ secure-context>> secure-context ] dip with-variable ; inline
 126
 127 : accept-loop ( server -- )
 128     [ accept-connection ] [ accept-loop ] bi ;
 129
 130 : start-accept-loop ( threaded-server server -- )
 131     '[ _ accept-loop ] with-existing-secure-context ;
 132
 133 \ start-accept-loop NOTICE add-error-logging
 134
 135 : create-secure-context ( threaded-server -- threaded-server )
 136     dup secure>> ssl-supported? and [
 137         dup secure-config>> <secure-context> >>secure-context
 138     ] when ;
 139
 140 : init-server ( threaded-server -- threaded-server )
 141     create-secure-context
 142     <flag> >>server-stopped
 143     dup semaphore>> [
 144         dup max-connections>> [
 145             <semaphore> >>semaphore
 146         ] when*
 147     ] unless ;
 148
 149 ERROR: no-ports-configured threaded-server ;
 150
 151 : (make-servers) ( theaded-server addrspecs -- servers )
 152     swap encoding>>
 153     '[ [ _ <server> |dispose ] map ] with-destructors ;
 154
 155 : set-servers ( threaded-server -- threaded-server )
 156     dup [
 157         dup dup listen-on
 158         [ no-ports-configured ] [ (make-servers) ] if-empty
 159         >>servers
 160     ] with-existing-secure-context ;
 161
 162 : server-thread-name ( threaded-server addrspec -- string )
 163     [ name>> ] [ addr>> present ] bi* " server on " glue ;
 164
 165 PRIVATE>
 166
 167 : start-server ( threaded-server -- threaded-server )
 168     init-server
 169     [
 170         dup threaded-server [
 171             [ ] [ name>> ] bi
 172             [
 173                 set-servers
 174                 dup running-servers get add-running-server
 175                 dup servers>>
 176                 [
 177                     [ '[ _ _ [ start-accept-loop ] with-disposal ] ]
 178                     [ server-thread-name ] 2bi spawn drop
 179                 ] with each
 180             ] with-logging
 181         ] with-variable
 182     ] keep ;
 183
 184 : server-running? ( threaded-server -- ? )
 185     server-stopped>> [ value>> not ] [ f ] if* ;
 186
 187 : stop-server ( threaded-server -- )
 188     dup server-running? [
 189         [ running-servers get remove-running-server ]
 190         [
 191             [
 192                 [ secure-context>> [ &dispose drop ] when* ]
 193                 [ [ f ] change-servers drop dispose-each ] bi
 194             ] with-destructors
 195         ]
 196         [ server-stopped>> raise-flag ] tri
 197     ] [
 198         drop
 199     ] if ;
 200
 201 : stop-this-server ( -- )
 202     threaded-server get stop-server ;
 203
 204 : wait-for-server ( threaded-server -- )
 205     server-stopped>> wait-for-flag ;
 206
 207 : with-threaded-server ( threaded-server quot -- )
 208     [ start-server ] dip over
 209     '[
 210         [ _ threaded-server _ with-variable ]
 211         [ _ stop-server ]
 212         finally
 213     ] call ; inline
 214
 215 <PRIVATE
 216
 217 GENERIC: connect-addr ( addrspec -- addrspec )
 218
 219 M: inet4 connect-addr [ "127.0.0.1" ] dip port>> <inet4> ;
 220
 221 M: inet6 connect-addr [ "::1" ] dip port>> <inet6> ;
 222
 223 M: secure connect-addr addrspec>> connect-addr f <secure> ;
 224
 225 M: local connect-addr ;
 226
 227 PRIVATE>
 228
 229 : server-addrs ( -- addrspecs )
 230     threaded-server get servers>> [ addr>> connect-addr ] map ;
 231
 232 : secure-addr ( -- addrspec )
 233     server-addrs [ secure? ] filter random ;
 234
 235 : insecure-addr ( -- addrspec )
 236     server-addrs [ secure? ] reject random ;
 237
 238 : server. ( threaded-server -- )
 239     [ [ "=== " write name>> ] [ ] bi write-object nl ]
 240     [ servers>> [ addr>> present print ] each ] bi ;
 241
 242 : all-servers ( -- sequence )
 243     running-servers get-global members ;
 244
 245 : get-servers-named ( string -- sequence )
 246     [ all-servers ] dip '[ name>> _ = ] filter ;
 247
 248 : servers. ( -- )
 249     all-servers [ server. ] each ;
 250
 251 : stop-all-servers ( -- )
 252     all-servers [ stop-server ] each ;