2
|
1 |
{-# LANGUAGE BangPatterns #-} |
|
2 |
|
|
3 |
module System.IO.Streams.Concurrent.Unagi.Bounded |
|
4 |
( -- * Channel conversions |
|
5 |
inputToChan |
|
6 |
, chanToInput |
|
7 |
, chanToOutput |
|
8 |
, makeChanPipe |
|
9 |
, dupStream |
|
10 |
, DupHandle |
|
11 |
) where |
|
12 |
|
|
13 |
|
|
14 |
------------------------------------------------------------------------------ |
|
15 |
import Control.Applicative (pure, (<$>), (<*>)) |
|
16 |
import Control.Concurrent.Chan.Unagi.Bounded (InChan, OutChan, |
|
17 |
dupChan, newChan, |
|
18 |
readChan, writeChan) |
|
19 |
import Control.Monad ((>=>)) |
|
20 |
import Prelude hiding (read) |
|
21 |
import System.IO.Streams.Internal (InputStream, |
|
22 |
OutputStream, |
|
23 |
makeInputStream, |
|
24 |
makeOutputStream, read) |
|
25 |
|
|
26 |
|
|
27 |
newtype DupHandle a = DupHandle { unDupHandle :: InChan (Maybe a) } |
|
28 |
|
|
29 |
------------------------------------------------------------------------------ |
|
30 |
-- | Writes the contents of an input stream to a channel until the input stream |
|
31 |
-- yields end-of-stream. |
|
32 |
inputToChan :: InputStream a -> InChan (Maybe a) -> IO () |
|
33 |
inputToChan is ch = go |
|
34 |
where |
|
35 |
go = do |
|
36 |
mb <- read is |
|
37 |
writeChan ch mb |
|
38 |
maybe (return $! ()) (const go) mb |
|
39 |
|
|
40 |
|
|
41 |
------------------------------------------------------------------------------ |
|
42 |
-- | Turns an 'OutChan' into an input stream. |
|
43 |
-- |
|
44 |
chanToInput :: OutChan (Maybe a) -> IO (InputStream a) |
|
45 |
chanToInput ch = makeInputStream $! readChan ch |
|
46 |
|
|
47 |
|
|
48 |
------------------------------------------------------------------------------ |
|
49 |
-- | Turns an 'InChan' into an output stream. |
|
50 |
-- |
|
51 |
chanToOutput :: InChan (Maybe a) -> IO (OutputStream a) |
|
52 |
chanToOutput = makeOutputStream . writeChan |
|
53 |
|
|
54 |
|
|
55 |
-------------------------------------------------------------------------------- |
|
56 |
-- | Create a new pair of streams using an underlying 'Chan'. Everything written |
|
57 |
-- to the 'OutputStream' will appear as-is on the 'InputStream'. |
|
58 |
-- |
|
59 |
-- Since reading from the 'InputStream' and writing to the 'OutputStream' are |
|
60 |
-- blocking calls, be sure to do so in different threads. |
|
61 |
makeChanPipe :: Int -> IO (InputStream a, OutputStream a, DupHandle a) |
|
62 |
makeChanPipe size = do |
|
63 |
(inChan, outChan) <- newChan size |
|
64 |
(,,) <$> chanToInput outChan <*> chanToOutput inChan <*> pure (DupHandle inChan) |
|
65 |
|
|
66 |
|
|
67 |
------------------------------------------------------------------------------ |
|
68 |
-- | Use a 'DupHandle' to replicate everything written on the |
|
69 |
-- associated 'OutputStream' to the 'InputStream'. |
|
70 |
-- |
|
71 |
dupStream :: DupHandle a -> IO (InputStream a) |
|
72 |
dupStream = dupChan . unDupHandle >=> chanToInput |