0d737d1 · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Update some handle and socket APIs

0d737d1Harendra Kumar7 years ago

src/Streamly/Internal/FileSystem/Handle.hs+11−12

compare with master

unifiedsplit

@@ -108,9 +108,6 @@ module Streamly.Internal.FileSystem.Handle

108where

109

110import Control.Monad.IO.Class (MonadIO(..))

111−#if __GLASGOW_HASKELL__ < 808

112−import Data.Semigroup (Semigroup(..))

113−#endif

114import Data.Word (Word8)

111import Data.Word (Word8)

115import Foreign.ForeignPtr (withForeignPtr)

112import Foreign.ForeignPtr (withForeignPtr)

116import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr)

113import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr)

@@ -133,7 +130,6 @@ import Streamly.Streams.StreamK.Type (IsStream, mkStream)

133

130

134import qualified Streamly.Data.Fold as FL

131import qualified Streamly.Data.Fold as FL

135import qualified Streamly.Internal.Data.Fold.Types as FL

132import qualified Streamly.Internal.Data.Fold.Types as FL

136−import qualified Streamly.Internal.Data.Unicode.Stream as U

137import qualified Streamly.Internal.Data.Unfold as UF

133import qualified Streamly.Internal.Data.Unfold as UF

138import qualified Streamly.Internal.Memory.Array as IA

134import qualified Streamly.Internal.Memory.Array as IA

139import qualified Streamly.Internal.Memory.ArrayStream as AS

135import qualified Streamly.Internal.Memory.ArrayStream as AS

@@ -367,26 +363,29 @@ fromChunks h m = S.mapM_ (liftIO . writeArray h) m

367putChunks :: (MonadIO m, Storable a) => SerialT m (Array a) -> m ()

363putChunks :: (MonadIO m, Storable a) => SerialT m (Array a) -> m ()

368putChunks = fromChunks stdout

364putChunks = fromChunks stdout

369

365

366+-- XXX use an unfold so that we can put any type of strings.

370−-- | Write a stream of strings to standard output using Latin1 encoding.

367+-- | Write a stream of strings to standard output using the supplied encoding.

371-- Output is flushed to the device for each string.

368-- Output is flushed to the device for each string.

372--

369--

373-- /Internal/

370-- /Internal/

374--

371--

375{-# INLINE putStrings #-}

372{-# INLINE putStrings #-}

376−putStrings :: MonadAsync m => SerialT m String -> m ()

373+putStrings :: MonadAsync m

374+ => (SerialT m Char -> SerialT m Word8) -> SerialT m String -> m ()

377−putStrings = putChunks . S.mapM (IA.fromStream . U.encodeLatin1 . S.fromList)

375+putStrings encode = putChunks . S.mapM (IA.fromStream . encode . S.fromList)

378

376

379-- XXX use an unfold so that we can put lines from any object

377-- XXX use an unfold so that we can put lines from any object

380−-- | Write a stream of strings as separate lines to standard output using

378+-- | Write a stream of strings as separate lines to standard output using the

381−-- Latin1 encoding. Output is line buffered i.e. the output is written to the

379+-- supplied encoding. Output is line buffered i.e. the output is written to the

382-- device as soon as a newline is encountered.

380-- device as soon as a newline is encountered.

383--

381--

384-- /Internal/

382-- /Internal/

385--

383--

386{-# INLINE putLines #-}

384{-# INLINE putLines #-}

387−putLines :: MonadAsync m => SerialT m String -> m ()

385+putLines :: MonadAsync m

386+ => (SerialT m Char -> SerialT m Word8) -> SerialT m String -> m ()

388−putLines = putChunks . S.mapM

387+putLines encode = putChunks . S.mapM

389− (\xs -> IA.fromStream $ U.encodeLatin1 (S.fromList xs <> S.yield '\n'))

388+ (\xs -> IA.fromStream $ encode (S.fromList (xs ++ "\n")))

390

389

391-- | Write a stream of bytes from standard output.

390-- | Write a stream of bytes from standard output.

392--

391--

src/Streamly/Internal/Network/Socket.hs+38−28

@@ -21,8 +21,8 @@ module Streamly.Internal.Network.Socket

21 (

22 SockSpec (..)

23 -- * Use a socket

24− , useSocketM

24+ , handleWithM

25− , useSocket

25+ , handleWith

27 -- * Accept connections

28 , accept

@@ -60,7 +60,7 @@ module Streamly.Internal.Network.Socket

60 , fromBytes

62 -- -- * Array Write

63− , writeArray

63+ , writeChunk

64 , writeChunks

65 , writeStrings

@@ -101,7 +101,6 @@ import Streamly.Data.Fold (Fold)

101-- import Streamly.String (encodeUtf8, decodeUtf8, foldLines)

102

103import qualified Streamly.Data.Fold as FL

104−import qualified Streamly.Data.Unicode.Stream as U

105import qualified Streamly.Internal.Data.Fold.Types as FL

104import qualified Streamly.Internal.Data.Fold.Types as FL

106import qualified Streamly.Internal.Data.Unfold as UF

105import qualified Streamly.Internal.Data.Unfold as UF

107import qualified Streamly.Internal.Memory.Array as IA

106import qualified Streamly.Internal.Memory.Array as IA

@@ -111,25 +110,25 @@ import qualified Streamly.Internal.Memory.Array.Types as A

111import qualified Streamly.Prelude as S

110import qualified Streamly.Prelude as S

112import qualified Streamly.Internal.Data.Stream.StreamD.Type as D

111import qualified Streamly.Internal.Data.Stream.StreamD.Type as D

113

112

114−-- | @'useSocketM' socket act@ runs the monadic computation @act@ passing the

113+-- | @'handleWithM' socket act@ runs the monadic computation @act@ passing the

115−-- socket handle to it. The handle will be closed on exit from 'useSocketM',

114+-- socket handle to it. The handle will be closed on exit from 'handleWithM',

116-- whether by normal termination or by raising an exception. If closing the

115-- whether by normal termination or by raising an exception. If closing the

117-- handle raises an exception, then this exception will be raised by

116-- handle raises an exception, then this exception will be raised by

118−-- 'useSocketM' rather than any exception raised by 'act'.

117+-- 'handleWithM' rather than any exception raised by 'act'.

119--

118--

120-- @since 0.7.0

119-- @since 0.7.0

121−{-# INLINE useSocketM #-}

120+{-# INLINE handleWithM #-}

122−useSocketM :: (MonadMask m, MonadIO m) => Socket -> (Socket -> m ()) -> m ()

121+handleWithM :: (MonadMask m, MonadIO m) => (Socket -> m ()) -> Socket -> m ()

123−useSocketM sk f = finally (f sk) (liftIO (Net.close sk))

122+handleWithM f sk = finally (f sk) (liftIO (Net.close sk))

124

123

125−-- | Like 'useSocketM' but runs a streaming computation instead of a monadic

124+-- | Like 'handleWithM' but runs a streaming computation instead of a monadic

126-- computation.

125-- computation.

127--

126--

128-- @since 0.7.0

127-- @since 0.7.0

129−{-# INLINE useSocket #-}

128+{-# INLINE handleWith #-}

130−useSocket :: (IsStream t, MonadCatch m, MonadIO m)

129+handleWith :: (IsStream t, MonadCatch m, MonadIO m)

131 => Socket -> (Socket -> t m a) -> t m a

130 => Socket -> (Socket -> t m a) -> t m a

132−useSocket sk f = S.finally (liftIO $ Net.close sk) (f sk)

131+handleWith sk f = S.finally (liftIO $ Net.close sk) (f sk)

133

132

134-------------------------------------------------------------------------------

133-------------------------------------------------------------------------------

135-- Accept (Unfolds)

134-- Accept (Unfolds)

@@ -282,19 +281,19 @@ writeArrayWith f h Array{..} = withForeignPtr aStart $ \p ->

282-- | Write an Array to a file handle.

281-- | Write an Array to a file handle.

283--

282--

284-- @since 0.7.0

283-- @since 0.7.0

285−{-# INLINABLE writeArray #-}

284+{-# INLINABLE writeChunk #-}

286−writeArray :: Storable a => Socket -> Array a -> IO ()

285+writeChunk :: Storable a => Socket -> Array a -> IO ()

287−writeArray = writeArrayWith sendAll

286+writeChunk = writeArrayWith sendAll

288

287

289-------------------------------------------------------------------------------

288-------------------------------------------------------------------------------

290-- Stream of Arrays IO

289-- Stream of Arrays IO

291-------------------------------------------------------------------------------

290-------------------------------------------------------------------------------

292

291

293−{-# INLINABLE readChunksUptoWith #-}

292+{-# INLINABLE _readChunksUptoWith #-}

294−readChunksUptoWith :: (IsStream t, MonadIO m)

293+_readChunksUptoWith :: (IsStream t, MonadIO m)

295 => (Int -> h -> IO (Array Word8))

294 => (Int -> h -> IO (Array Word8))

296 -> Int -> h -> t m (Array Word8)

295 -> Int -> h -> t m (Array Word8)

297−readChunksUptoWith f size h = go

296+_readChunksUptoWith f size h = go

298 where

297 where

299 -- XXX use cons/nil instead

298 -- XXX use cons/nil instead

300 go = mkStream $ \_ yld _ stp -> do

299 go = mkStream $ \_ yld _ stp -> do

@@ -307,10 +306,19 @@ readChunksUptoWith f size h = go

307-- The maximum size of a single array is limited to @size@.

306-- The maximum size of a single array is limited to @size@.

308-- 'fromHandleArraysUpto' ignores the prevailing 'TextEncoding' and 'NewlineMode'

307-- 'fromHandleArraysUpto' ignores the prevailing 'TextEncoding' and 'NewlineMode'

309-- on the 'Handle'.

308-- on the 'Handle'.

310−{-# INLINABLE toChunksWithBufferOf #-}

309+{-# INLINE_NORMAL toChunksWithBufferOf #-}

311toChunksWithBufferOf :: (IsStream t, MonadIO m)

310toChunksWithBufferOf :: (IsStream t, MonadIO m)

312 => Int -> Socket -> t m (Array Word8)

311 => Int -> Socket -> t m (Array Word8)

312+-- toChunksWithBufferOf = _readChunksUptoWith readArrayOf

313−toChunksWithBufferOf = readChunksUptoWith readArrayOf

313+toChunksWithBufferOf size h = D.fromStreamD (D.Stream step ())

314+ where

315+ {-# INLINE_LATE step #-}

316+ step _ _ = do

317+ arr <- liftIO $ readArrayOf size h

318+ return $

319+ case A.length arr of

320+ 0 -> D.Stop

321+ _ -> D.Yield arr ()

314

322

315-- XXX read 'Array a' instead of Word8

323-- XXX read 'Array a' instead of Word8

316--

324--

@@ -408,7 +416,7 @@ read = UF.supplyFirst readWithBufferOf A.defaultChunkSize

408{-# INLINE fromChunks #-}

416{-# INLINE fromChunks #-}

409fromChunks :: (MonadIO m, Storable a)

417fromChunks :: (MonadIO m, Storable a)

410 => Socket -> SerialT m (Array a) -> m ()

418 => Socket -> SerialT m (Array a) -> m ()

411−fromChunks h m = S.mapM_ (liftIO . writeArray h) m

419+fromChunks h m = S.mapM_ (liftIO . writeChunk h) m

412

420

413-- | Write a stream of arrays to a socket. Each array in the stream is written

421-- | Write a stream of arrays to a socket. Each array in the stream is written

414-- to the socket as a separate IO request.

422-- to the socket as a separate IO request.

@@ -416,16 +424,18 @@ fromChunks h m = S.mapM_ (liftIO . writeArray h) m

416-- @since 0.7.0

424-- @since 0.7.0

417{-# INLINE writeChunks #-}

425{-# INLINE writeChunks #-}

418writeChunks :: (MonadIO m, Storable a) => Socket -> Fold m (Array a) ()

426writeChunks :: (MonadIO m, Storable a) => Socket -> Fold m (Array a) ()

419−writeChunks h = FL.drainBy (liftIO . writeArray h)

427+writeChunks h = FL.drainBy (liftIO . writeChunk h)

420

428

421−-- | Write a stream of strings to a socket in Latin1 encoding.

429+-- | Write a stream of strings to a socket in Latin1 encoding. Output is

430+-- flushed to the socket for each string.

422--

431--

423-- /Internal/

432-- /Internal/

424--

433--

425{-# INLINE writeStrings #-}

434{-# INLINE writeStrings #-}

426−writeStrings :: MonadIO m => Socket -> Fold m String ()

435+writeStrings :: MonadIO m

436+ => (SerialT m Char -> SerialT m Word8) -> Socket -> Fold m String ()

427−writeStrings h =

437+writeStrings encode h =

428− FL.lmapM (IA.fromStream . U.encodeLatin1 . S.fromList) (writeChunks h)

438+ FL.lmapM (IA.fromStream . encode . S.fromList) (writeChunks h)

429

439

430-- GHC buffer size dEFAULT_FD_BUFFER_SIZE=8192 bytes.

440-- GHC buffer size dEFAULT_FD_BUFFER_SIZE=8192 bytes.

431--

441--