git.iko.soy/streamly
on master help sign in
Add transformBytesWith to transform a stream via a server
commit23e00d0c34078c58e5f3c8f121e754474cef8a0f authorHarendra Kumar <harendra.kumar@gmail.com> dateThu 28 Nov 2019, 07:25 (7 years ago) parentba6a1e7
2 files changed+56−3
compare with master
unifiedsplit
src/Streamly/Internal/Data/SVar.hs+10−3 src/Streamly/Internal/Network/Inet/TCP.hs+46
src/Streamly/Internal/Data/SVar.hs+10−3
@@ -103,6 +103,7 @@ module Streamly.Internal.Data.SVar
103103 , delThread
104104 , modifyThread
105105 , doFork
106+ , fork
106107
107108 , toStreamVar
108109 , SVarStats (..)
@@ -112,9 +113,10 @@ module Streamly.Internal.Data.SVar
112113where
113114
114115import Control.Concurrent
115− (ThreadId, myThreadId, threadDelay, throwTo)
116+ (ThreadId, myThreadId, threadDelay, throwTo, forkIO)
116117import Control.Concurrent.MVar
117− (MVar, newEmptyMVar, tryPutMVar, takeMVar, tryTakeMVar, newMVar, tryReadMVar)
118+ (MVar, newEmptyMVar, tryPutMVar, takeMVar, tryTakeMVar, newMVar,
119+ tryReadMVar)
118120import Control.Exception
119121 (SomeException(..), catch, mask, assert, Exception, catches,
120122 throwIO, Handler(..), BlockedIndefinitelyOnMVar(..),
@@ -122,7 +124,8 @@ import Control.Exception
122124import Control.Monad (when)
123125import Control.Monad.Catch (MonadThrow)
124126import 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)
126129import Streamly.Internal.Data.Atomics
127130 (atomicModifyIORefCAS, atomicModifyIORefCAS_, writeBarrier,
128131 storeLoadBarrier)
@@ -939,6 +942,10 @@ doFork action (RunInIO mrun) exHandler =
939942 exHandler
940943 run (return tid)
941944
945+{-# INLINABLE fork #-}
946+fork :: MonadBaseControl IO m => m () -> m ThreadId
947+fork = liftBaseDiscard forkIO
948+
942949------------------------------------------------------------------------------
943950-- Collecting results from child workers in a streamed fashion
944951------------------------------------------------------------------------------
src/Streamly/Internal/Network/Inet/TCP.hs+46
@@ -71,6 +71,9 @@ module Streamly.Internal.Network.Inet.TCP
7171 -- , writeArray
7272 , writeChunks
7373 , fromChunks
74+
75+ -- ** Transformation
76+ , transformBytesWith
7477 {-
7578 -- ** Sink Servers
7679
@@ -106,6 +109,7 @@ import Prelude hiding (read)
106109
107110import Streamly (MonadAsync)
108111import Streamly.Internal.Data.Fold.Types (Fold(..))
112+import Streamly.Internal.Data.SVar (fork)
109113import Streamly.Internal.Data.Unfold.Types (Unfold(..))
110114import Streamly.Internal.Network.Socket (SockSpec(..), accept, connections)
111115import Streamly.Streams.Serial (SerialT)
@@ -386,3 +390,45 @@ fromBytes = fromBytesWithBufferOf defaultChunkSize
386390write :: (MonadAsync m, MonadCatch m)
387391 => (Word8, Word8, Word8, Word8) -> PortNumber -> Fold m Word8 ()
388392write = 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