git.iko.soy/streamly
on master help sign in
Add transformBytesWith to transform a stream via a server
23e00d0Harendra Kumar
src/Streamly/Internal/Data/SVar.hs+10−3
compare with master
unifiedsplit
@@ -103,6 +103,7 @@ module Streamly.Internal.Data.SVar
103 , delThread
103 , delThread
104 , modifyThread
104 , modifyThread
105 , doFork
105 , doFork
106+ , fork
106
107
107 , toStreamVar
108 , toStreamVar
108 , SVarStats (..)
109 , SVarStats (..)
@@ -112,9 +113,10 @@ module Streamly.Internal.Data.SVar
112where
113where
113
114
114import Control.Concurrent
115import Control.Concurrent
115− (ThreadId, myThreadId, threadDelay, throwTo)
116+ (ThreadId, myThreadId, threadDelay, throwTo, forkIO)
116import Control.Concurrent.MVar
117import Control.Concurrent.MVar
117− (MVar, newEmptyMVar, tryPutMVar, takeMVar, tryTakeMVar, newMVar, tryReadMVar)
118+ (MVar, newEmptyMVar, tryPutMVar, takeMVar, tryTakeMVar, newMVar,
119+ tryReadMVar)
118import Control.Exception
120import Control.Exception
119 (SomeException(..), catch, mask, assert, Exception, catches,
121 (SomeException(..), catch, mask, assert, Exception, catches,
120 throwIO, Handler(..), BlockedIndefinitelyOnMVar(..),
122 throwIO, Handler(..), BlockedIndefinitelyOnMVar(..),
@@ -122,7 +124,8 @@ import Control.Exception
122import Control.Monad (when)
124import Control.Monad (when)
123import Control.Monad.Catch (MonadThrow)
125import Control.Monad.Catch (MonadThrow)
124import Control.Monad.IO.Class (MonadIO(..))
126import Control.Monad.IO.Class (MonadIO(..))
125−import Control.Monad.Trans.Control (MonadBaseControl, control, StM)
127+import Control.Monad.Trans.Control
128+ (MonadBaseControl, control, StM, liftBaseDiscard)
126import Streamly.Internal.Data.Atomics
129import Streamly.Internal.Data.Atomics
127 (atomicModifyIORefCAS, atomicModifyIORefCAS_, writeBarrier,
130 (atomicModifyIORefCAS, atomicModifyIORefCAS_, writeBarrier,
128 storeLoadBarrier)
131 storeLoadBarrier)
@@ -939,6 +942,10 @@ doFork action (RunInIO mrun) exHandler =
939 exHandler
942 exHandler
940 run (return tid)
943 run (return tid)
941
944
945+{-# INLINABLE fork #-}
946+fork :: MonadBaseControl IO m => m () -> m ThreadId
947+fork = liftBaseDiscard forkIO
948+
942------------------------------------------------------------------------------
949------------------------------------------------------------------------------
943-- Collecting results from child workers in a streamed fashion
950-- Collecting results from child workers in a streamed fashion
944------------------------------------------------------------------------------
951------------------------------------------------------------------------------
src/Streamly/Internal/Network/Inet/TCP.hs+46
@@ -71,6 +71,9 @@ module Streamly.Internal.Network.Inet.TCP
71 -- , writeArray
71 -- , writeArray
72 , writeChunks
72 , writeChunks
73 , fromChunks
73 , fromChunks
74+
75+ -- ** Transformation
76+ , transformBytesWith
74 {-
77 {-
75 -- ** Sink Servers
78 -- ** Sink Servers
76
79
@@ -106,6 +109,7 @@ import Prelude hiding (read)
106
109
107import Streamly (MonadAsync)
110import Streamly (MonadAsync)
108import Streamly.Internal.Data.Fold.Types (Fold(..))
111import Streamly.Internal.Data.Fold.Types (Fold(..))
112+import Streamly.Internal.Data.SVar (fork)
109import Streamly.Internal.Data.Unfold.Types (Unfold(..))
113import Streamly.Internal.Data.Unfold.Types (Unfold(..))
110import Streamly.Internal.Network.Socket (SockSpec(..), accept, connections)
114import Streamly.Internal.Network.Socket (SockSpec(..), accept, connections)
111import Streamly.Streams.Serial (SerialT)
115import Streamly.Streams.Serial (SerialT)
@@ -386,3 +390,45 @@ fromBytes = fromBytesWithBufferOf defaultChunkSize
386write :: (MonadAsync m, MonadCatch m)
390write :: (MonadAsync m, MonadCatch m)
387 => (Word8, Word8, Word8, Word8) -> PortNumber -> Fold m Word8 ()
391 => (Word8, Word8, Word8, Word8) -> PortNumber -> Fold m Word8 ()
388write = writeWithBufferOf defaultChunkSize
392write = writeWithBufferOf defaultChunkSize
393+
394+-------------------------------------------------------------------------------
395+-- Transformations
396+-------------------------------------------------------------------------------
397+
398+{-# INLINABLE withInputConnect #-}
399+withInputConnect
400+ :: (IsStream t, MonadCatch m, MonadAsync m)
401+ => (Word8, Word8, Word8, Word8)
402+ -> PortNumber
403+ -> SerialT m Word8
404+ -> (Socket -> t m a)
405+ -> t m a
406+withInputConnect addr port input f = S.bracket pre post handler
407+
408+ where
409+
410+ pre = do
411+ sk <- liftIO $ connect addr port
412+ tid <- fork (ISK.fromBytes sk input)
413+ return (sk, tid)
414+
415+ handler (sk, _) = f sk
416+
417+ -- XXX kill the thread immediately?
418+ post (sk, _) = liftIO $ Net.close sk
419+
420+-- | Send an input stream to a remote host and produce the output stream from
421+-- the host. The server host just acts as a transformation function on the
422+-- input stream. Both sending and receiving happen asynchronously.
423+--
424+-- /Internal/
425+--
426+{-# INLINABLE transformBytesWith #-}
427+transformBytesWith
428+ :: (IsStream t, MonadAsync m, MonadCatch m)
429+ => (Word8, Word8, Word8, Word8)
430+ -> PortNumber
431+ -> SerialT m Word8
432+ -> t m Word8
433+transformBytesWith addr port input =
434+ withInputConnect addr port input ISK.toBytes