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