extra/io/nonblocking/nonblocking.factor

   1 ! Copyright (C) 2007 Slava Pestov, Doug Coleman
   2 ! See http://factorcode.org/license.txt for BSD license.
   3 IN: io.nonblocking
   4 USING: math kernel io sequences io.buffers generic sbufs
   5 system io.streams.lines io.streams.plain io.streams.duplex
   6 continuations debugger classes byte-arrays ;
   7
   8 : default-buffer-size 64 1024 * ; inline
   9
  10 ! Common delegate of native stream readers and writers
  11 TUPLE: port handle error timeout cutoff type eof? ;
  12
  13 SYMBOL: input
  14 SYMBOL: output
  15 SYMBOL: closed
  16
  17 PREDICATE: port input-port port-type input eq? ;
  18 PREDICATE: port output-port port-type output eq? ;
  19
  20 GENERIC: init-handle ( handle -- )
  21
  22 : <port> ( handle buffer -- port )
  23     over init-handle
  24     0 0 {
  25         set-port-handle
  26         set-delegate
  27         set-port-timeout
  28         set-port-cutoff
  29     } port construct ;
  30
  31 : <buffered-port> ( handle -- port )
  32     default-buffer-size <buffer> <port> ;
  33
  34 : <reader> ( handle -- stream )
  35     <buffered-port> input over set-port-type <line-reader> ;
  36
  37 : <writer> ( handle -- stream )
  38     <buffered-port> output over set-port-type <plain-writer> ;
  39
  40 : handle>duplex-stream ( in-handle out-handle -- stream )
  41     <writer>
  42     [ >r <reader> r> <duplex-stream> ]
  43     [ ] [ stream-close ]
  44     cleanup ;
  45
  46 : touch-port ( port -- )
  47     dup port-timeout dup zero?
  48     [ 2drop ] [ millis + swap set-port-cutoff ] if ;
  49
  50 : timeout? ( port -- ? )
  51     port-cutoff dup zero? not swap millis < and ;
  52
  53 : pending-error ( port -- )
  54     dup port-error f rot set-port-error [ throw ] when* ;
  55
  56 M: port set-timeout
  57     [ set-port-timeout ] keep touch-port ;
  58
  59 GENERIC: (wait-to-read) ( port -- )
  60
  61 : wait-to-read ( count port -- )
  62     tuck buffer-length > [ (wait-to-read) ] [ drop ] if ;
  63
  64 : wait-to-read1 ( port -- )
  65     1 swap wait-to-read ;
  66
  67 : unless-eof ( port quot -- value )
  68     >r dup buffer-empty? over port-eof? and
  69     [ f swap set-port-eof? f ] r> if ; inline
  70
  71 M: input-port stream-read1
  72     dup wait-to-read1 [ buffer-pop ] unless-eof ;
  73
  74 : read-step ( count port -- string/f )
  75     [ wait-to-read ] 2keep
  76     [ dupd buffer> ] unless-eof nip ;
  77
  78 : read-loop ( count port sbuf -- )
  79     pick over length - dup 0 > [
  80         pick read-step dup [
  81             over push-all read-loop
  82         ] [
  83             2drop 2drop
  84         ] if
  85     ] [
  86         2drop 2drop
  87     ] if ;
  88
  89 M: input-port stream-read
  90     >r 0 max >fixnum r>
  91     2dup read-step dup [
  92         pick over length > [
  93             pick <sbuf>
  94             [ push-all ] keep
  95             [ read-loop ] keep
  96             "" like
  97         ] [
  98             2nip
  99         ] if
 100     ] [
 101         2nip
 102     ] if ;
 103
 104 : read-until-step ( separators port -- string/f separator/f )
 105     dup wait-to-read1
 106     dup port-eof? [
 107         f swap set-port-eof? drop f f
 108     ] [
 109         buffer-until
 110     ] if ;
 111
 112 : read-until-loop ( seps port sbuf -- separator/f )
 113     pick pick read-until-step over [
 114         >r over push-all r> dup [
 115             >r 3drop r>
 116         ] [
 117             drop read-until-loop
 118         ] if
 119     ] [
 120         >r 2drop 2drop r>
 121     ] if ;
 122
 123 M: input-port stream-read-until ( seps port -- str/f sep/f )
 124     2dup read-until-step dup [
 125         >r 2nip r>
 126     ] [
 127         over [
 128             drop >sbuf [ read-until-loop ] keep "" like swap
 129         ] [
 130             >r 2nip r>
 131         ] if
 132     ] if ;
 133
 134 M: input-port stream-read-partial ( max stream -- string/f )
 135     >r 0 max >fixnum r> read-step ;
 136
 137 : can-write? ( len writer -- ? )
 138     dup buffer-empty? [
 139         2drop t
 140     ] [
 141         [ buffer-fill + ] keep buffer-capacity <=
 142     ] if ;
 143
 144 : wait-to-write ( len port -- )
 145     tuck can-write? [ drop ] [ stream-flush ] if ;
 146
 147 M: output-port stream-write1
 148     1 over wait-to-write ch>buffer ;
 149
 150 M: output-port stream-write
 151     over length over wait-to-write >buffer ;
 152
 153 TUPLE: server-port addr client ;
 154
 155 : <server-port> ( port addr -- server )
 156     server-port pick set-port-type
 157     { set-delegate set-server-port-addr }
 158     server-port construct ;
 159
 160 : check-server-port ( port -- )
 161     port-type server-port assert= ;
 162
 163 TUPLE: datagram-port addr packet packet-addr ;
 164
 165 : <datagram-port> ( port addr -- datagram )
 166     datagram-port pick set-port-type
 167     { set-delegate set-datagram-port-addr }
 168     datagram-port construct ;
 169
 170 : check-datagram-port ( port -- )
 171     port-type datagram-port assert= ;
 172
 173 : check-datagram-send ( packet addrspec port -- )
 174     dup check-datagram-port
 175     datagram-port-addr [ class ] 2apply assert=
 176     class byte-array assert= ;