basis/io/servers/connection/connection.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.connection
  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
  27 SYMBOL: running-servers
  28 running-servers [ HS{ } clone ] initialize
  29
  30 ERROR: server-already-running threaded-server ;
  31
  32 ERROR: server-not-running threaded-server ;
  33
  34 <PRIVATE
  35
  36 : must-be-running ( threaded-server -- threaded-server )
  37     dup running-servers get in? [ server-not-running ] unless ;
  38
  39 : must-not-be-running ( threaded-server -- threaded-server )
  40     dup running-servers get in? [ server-already-running ] when ;
  41
  42 : add-running-server ( threaded-server -- )
  43     must-not-be-running
  44     running-servers get adjoin ;
  45
  46 : remove-running-server ( threaded-server -- )
  47     ! must-be-running
  48     running-servers get delete ;
  49
  50 PRIVATE>
  51
  52 : local-server ( port -- addrspec ) "localhost" swap <inet> ;
  53
  54 : internet-server ( port -- addrspec ) f swap <inet> ;
  55
  56 : new-threaded-server ( encoding class -- threaded-server )
  57     new
  58         "server" >>name
  59         DEBUG >>log-level
  60         <secure-config> >>secure-config
  61         1 minutes >>timeout
  62         [ "No handler quotation" throw ] >>handler
  63         swap >>encoding ;
  64
  65 : <threaded-server> ( encoding -- threaded-server )
  66     threaded-server new-threaded-server ;
  67
  68 GENERIC: handle-client* ( threaded-server -- )
  69
  70 <PRIVATE
  71
  72 GENERIC: (>insecure) ( obj -- obj )
  73
  74 M: inet (>insecure) ;
  75 M: inet4 (>insecure) ;
  76 M: inet6 (>insecure) ;
  77 M: local (>insecure) ;
  78 M: integer (>insecure) internet-server ;
  79 M: string (>insecure) internet-server ;
  80 M: array (>insecure) [ (>insecure) ] map ;
  81 M: f (>insecure) ;
  82
  83 : >insecure ( obj -- seq )
  84     (>insecure) dup sequence? [ 1array ] unless ;
  85
  86 : >secure ( addrspec -- addrspec' )
  87     >insecure
  88     [ dup { [ secure? ] [ not ] } 1|| [ <secure> ] unless ] map ;
  89
  90 : listen-on ( threaded-server -- addrspecs )
  91     [ secure>> >secure ] [ insecure>> >insecure ] 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 : accept-loop ( server -- )
 135     [ accept-connection ] [ accept-loop ] bi ;
 136
 137 : start-accept-loop ( server -- ) accept-loop ;
 138
 139 \ start-accept-loop NOTICE add-error-logging
 140
 141 : init-server ( threaded-server -- threaded-server )
 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 dup listen-on [ no-ports-configured ] [ (make-servers) ] if-empty
 157     >>servers ;
 158
 159 : server-thread-name ( threaded-server addrspec -- string )
 160     [ name>> ] [ addr>> present ] bi* " server on " glue ;
 161
 162 : (start-server) ( threaded-server -- )
 163     init-server
 164     dup threaded-server [
 165         [ ] [ name>> ] bi
 166         [
 167             set-servers
 168             dup add-running-server
 169             dup servers>>
 170             [
 171                 [ nip '[ _ [ start-accept-loop ] with-disposal ] ]
 172                 [ server-thread-name ] 2bi spawn drop
 173             ] with each
 174         ] with-logging
 175     ] with-variable ;
 176
 177 PRIVATE>
 178
 179 : start-server ( threaded-server -- threaded-server )
 180     #! Only create a secure-context if we want to listen on
 181     #! a secure port, otherwise start-server won't work at
 182     #! all if SSL is not available.
 183     dup dup secure>> [
 184         dup secure-config>> [
 185             (start-server)
 186         ] with-secure-context
 187     ] [
 188         (start-server)
 189     ] if ;
 190
 191 : server-running? ( threaded-server -- ? )
 192     server-stopped>> [ value>> not ] [ f ] if* ;
 193
 194 : stop-server ( threaded-server -- )
 195     dup server-running? [
 196         [ [ f ] change-servers drop dispose-each ]
 197         [ remove-running-server ]
 198         [ server-stopped>> raise-flag ] tri
 199     ] [
 200         drop
 201     ] if ;
 202
 203 : stop-this-server ( -- )
 204     threaded-server get stop-server ;
 205
 206 : wait-for-server ( threaded-server -- )
 207     server-stopped>> wait-for-flag ;
 208
 209 : with-threaded-server ( threaded-server quot -- )
 210     over
 211     '[
 212         [ _ start-server threaded-server _ with-variable ]
 213         [ _ stop-server ]
 214         [ ] cleanup
 215     ] call ; inline
 216
 217 <PRIVATE
 218
 219 : first-port ( quot -- n/f )
 220     [ threaded-server get servers>> ] dip
 221     filter [ f ] [ first addr>> port>> ] if-empty ; inline
 222
 223 PRIVATE>
 224
 225 : secure-port ( -- n/f ) [ addr>> secure? ] first-port ;
 226
 227 : insecure-port ( -- n/f ) [ addr>> secure? not ] first-port ;
 228
 229 : secure-addr ( -- inet )
 230     threaded-server get servers>> [ addr>> secure? ] filter random ;
 231
 232 : insecure-addr ( -- inet )
 233     threaded-server get servers>> [ addr>> secure? not ] filter random addr>> ;
 234
 235 : server. ( threaded-server -- )
 236     [ [ "=== " write name>> ] [ ] bi write-object nl ]
 237     [ servers>> [ addr>> present print ] each ] bi ;
 238
 239 : all-servers ( -- sequence )
 240     running-servers get-global members ;
 241
 242 : get-servers-named ( string -- sequence )
 243     [ all-servers ] dip '[ name>> _ = ] filter ;
 244
 245 : servers. ( -- )
 246     all-servers [ server. ] each ;
 247
 248 : stop-all-servers ( -- )
 249     all-servers [ stop-server ] each ;