basis/io/pipes/pipes.factor

   1 ! Copyright (C) 2008, 2009 Slava Pestov.
   2 ! See http://factorcode.org/license.txt for BSD license.
   3 USING: io.encodings io.backend io.ports io.streams.duplex
   4 io splitting grouping sequences namespaces kernel
   5 destructors math concurrency.combinators accessors fry
   6 arrays continuations quotations system vocabs.loader combinators ;
   7 IN: io.pipes
   8
   9 TUPLE: pipe in out ;
  10
  11 M: pipe dispose ( pipe -- )
  12     [ in>> dispose ] [ out>> dispose ] bi ;
  13
  14 HOOK: (pipe) io-backend ( -- pipe )
  15
  16 : <pipe> ( encoding -- stream )
  17     [
  18         [
  19             (pipe) |dispose
  20             [ in>> <input-port> ] [ out>> <output-port> ] bi
  21         ] dip <encoder-duplex>
  22     ] with-destructors ;
  23
  24 <PRIVATE
  25
  26 : ?reader ( handle/f -- stream )
  27     [ <input-port> &dispose ] [ input-stream get ] if* ;
  28
  29 : ?writer ( handle/f -- stream )
  30     [ <output-port> &dispose ] [ output-stream get ] if* ;
  31
  32 GENERIC: run-pipeline-element ( input-fd output-fd obj -- result )
  33
  34 M: callable run-pipeline-element
  35     [
  36         [ [ ?reader ] [ ?writer ] bi* ] dip
  37         '[ _ call( -- result ) ] with-streams*
  38     ] with-destructors ;
  39
  40 GENERIC: <pipes> ( obj -- pipes )
  41
  42 M: integer <pipes> ( n -- pipes )
  43     [
  44         [ (pipe) |dispose ] replicate
  45         T{ pipe } [ prefix ] [ suffix ] bi
  46         2 <clumps>
  47     ] with-destructors ;
  48
  49 M: sequence <pipes>
  50     [ { } ] [ length 1 - <pipes> ] if-empty ;
  51
  52 PRIVATE>
  53
  54 : run-pipeline ( seq -- results )
  55     [ <pipes> ] keep
  56     [
  57         [ [ first in>> ] [ second out>> ] bi ] dip
  58         run-pipeline-element
  59     ] 2parallel-map ;
  60
  61 {
  62     { [ os unix? ] [ "io.pipes.unix" require ] }
  63     { [ os windows? ] [ "io.pipes.windows" require ] }
  64     [ ]
  65 } cond