stream (empty) → 0.1.0.0
raw patch · 10 files changed
+517/−0 lines, 10 filesdep +basedep +bytestringdep +criterionsetup-changed
Dependencies added: base, bytestring, criterion, exceptions, ghc-prim, hspec, mtl, stream, streaming-commons, temporary, transformers, vector
Files
- ChangeLog.md +3/−0
- LICENSE +20/−0
- README.md +13/−0
- Setup.hs +2/−0
- bench/stream-bench.hs +68/−0
- src/Stream.hs +227/−0
- src/Stream/Core.hs +83/−0
- src/Stream/Core/Internal.hs +13/−0
- stream.cabal +57/−0
- test/Spec.hs +31/−0
+ ChangeLog.md view
@@ -0,0 +1,3 @@+## 0.1.0.0++* Initial pre-release version, not at all ready for usage!
+ LICENSE view
@@ -0,0 +1,20 @@+Copyright (c) 2016 FP Complete, https://www.fpcomplete.com/++Permission is hereby granted, free of charge, to any person obtaining+a copy of this software and associated documentation files (the+"Software"), to deal in the Software without restriction, including+without limitation the rights to use, copy, modify, merge, publish,+distribute, sublicense, and/or sell copies of the Software, and to+permit persons to whom the Software is furnished to do so, subject to+the following conditions:++The above copyright notice and this permission notice shall be+included in all copies or substantial portions of the Software.++THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,+EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF+MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND+NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE+LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION+OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION+WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+ README.md view
@@ -0,0 +1,13 @@+## stream++[](https://travis-ci.org/fpco/stream)++*NOTE* This library is very much in prerelease state. If you see this released+anywhere, it's just available for easier sharing with others. Please do _not_+start using it yet.++Streaming data library built around making stream fusion a first-class concept.+Focus is on high performance and usability.++Unlike more commonly used streaming libraries (like conduit, enumerator, or+pipes), this library is _not_ coroutine based.
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ bench/stream-bench.hs view
@@ -0,0 +1,68 @@+{-# LANGUAGE BangPatterns #-}+{-# LANGUAGE MagicHash #-}+import Criterion.Main+import Stream+import qualified Data.Vector as VB+import qualified Data.Vector.Unboxed as VU+import GHC.Prim+import GHC.Types+import System.IO (hClose)+import System.IO.Temp (withSystemTempFile)+import qualified Data.ByteString as S+import qualified Data.ByteString.Lazy as L+import Data.ByteString.Lazy.Internal (defaultChunkSize)+import qualified System.IO as IO++withTempFP :: String -> (FilePath -> IO a) -> IO a+withTempFP str inner = withSystemTempFile str $ \fp h -> do+ hClose h+ inner fp++main :: IO ()+main = withTempFP "src" $ \srcFP -> withTempFP "dst" $ \dstFP -> do+ L.writeFile dstFP $ L.fromChunks $ replicate 10000 $ S.replicate 1000 65++ defaultMain+ [ let high = 1000000 :: Int+ go name f = bench name $ whnf f high+ {-# INLINE go #-}+ in bgroup "enum/map/sum"+ [ go "stream" $ runIdentity . sumS . mapS (+ 1) . enumFromToS 1+ , go "prim" $ \(I# high') ->+ let loop x total =+ case x +# 1# of+ y ->+ case total +# y of+ total' ->+ if isTrue# (x <=# high')+ then loop y total'+ else I# total'+ in loop 1# 0#+ , go "low level" $ \high' ->+ let loop !x !total+ | x <= high' = loop y total'+ | otherwise = total'+ where+ !y = x + 1+ !total' = total + y+ {-# INLINE loop #-}+ in loop 1 0+ , go "boxed vector" $ VB.sum . VB.map (+ 1) . VB.enumFromTo 1+ , go "unboxed vector" $ VU.sum . VU.map (+ 1) . VU.enumFromTo 1+ ]+ , bgroup "file copy"+ [ bench "stream" $ whnfIO $ writeFileS dstFP $ readFileS srcFP+ , bench "lazy I/O" $ whnfIO $ L.readFile srcFP >>= L.writeFile dstFP+ , bench "low level" $ whnfIO $+ IO.withBinaryFile srcFP IO.ReadMode $ \src ->+ IO.withBinaryFile dstFP IO.WriteMode $ \dst ->+ let loop = do+ bs <- S.hGetSome src defaultChunkSize+ if S.null bs+ then return ()+ else do+ S.hPut dst bs+ loop+ in loop+ ]+ ]
+ src/Stream.hs view
@@ -0,0 +1,227 @@+{-# LANGUAGE BangPatterns #-}+module Stream+ ( -- * Types+ StreamT+ -- * Common+ , runStreamT++ -- * Sources+ -- ** Pure+ , enumFromToS+ , yieldS++ -- * Transformers+ -- ** Pure+ , mapS+ -- ** Monadic++ -- * Sinks+ -- ** Pure+ , foldlS+ , sumS+ , sinkListS+ -- ** Monadic+ , mapM_S++ -- * I/O+ , readFileS+ , writeFileS++ -- * Textual+ , linesAsciiS++ -- * Reexports+ , Identity (..)+ ) where++import Stream.Core+import Data.Functor.Identity (Identity (..))+import Data.Int+import Data.Word+import qualified Control.Monad.Catch as Catch+import Control.Monad.IO.Class (MonadIO, liftIO)+import qualified Data.ByteString as S+import qualified System.IO as IO+import qualified Data.Streaming.FileRead as FR+import qualified Data.Foldable as F++enumFromToS :: (Ord a, Enum a, Applicative m)+ => a+ -> a+ -> StreamT a m ()+enumFromToS start end =+ makeSource start go+ where+ go x+ | x <= end =+ let !y = succ x+ in pure (Yield y x)+ | otherwise = pure (Done ())+{-# INLINE [1] enumFromToS #-}++enumFromToS_num+ :: (Ord a, Num a, Applicative m)+ => a+ -> a+ -> StreamT a m ()+enumFromToS_num start end =+ makeSource start go+ where+ go x+ | x <= end =+ let !y = x + 1+ in pure (Yield y x)+ | otherwise = pure (Done ())+{-# INLINE [0] enumFromToS_num #-}+{-# RULES++"enumFromToS<Int>" enumFromToS =+ enumFromToS_num :: Applicative m => Int -> Int -> StreamT Int m ()++"enumFromToS<Int8>" enumFromToS =+ enumFromToS_num :: Applicative m => Int8 -> Int8 -> StreamT Int8 m ()++"enumFromToS<Int16>" enumFromToS =+ enumFromToS_num :: Applicative m => Int16 -> Int16 -> StreamT Int16 m ()++"enumFromToS<Int32>" enumFromToS =+ enumFromToS_num :: Applicative m => Int32 -> Int32 -> StreamT Int32 m ()++"enumFromToS<Int64>" enumFromToS =+ enumFromToS_num :: Applicative m => Int64 -> Int64 -> StreamT Int64 m ()++"enumFromToS<Word>" enumFromToS =+ enumFromToS_num :: Applicative m => Word -> Word -> StreamT Word m ()++"enumFromToS<Word8>" enumFromToS =+ enumFromToS_num :: Applicative m => Word8 -> Word8 -> StreamT Word8 m ()++"enumFromToS<Word16>" enumFromToS =+ enumFromToS_num :: Applicative m => Word16 -> Word16 -> StreamT Word16 m ()++"enumFromToS<Word32>" enumFromToS =+ enumFromToS_num :: Applicative m => Word32 -> Word32 -> StreamT Word32 m ()++"enumFromToS<Word64>" enumFromToS =+ enumFromToS_num :: Applicative m => Word64 -> Word64 -> StreamT Word64 m ()++"enumFromToS<Integer>" enumFromToS =+ enumFromToS_num :: Applicative m => Integer -> Integer -> StreamT Integer m () #-}++yieldS :: (F.Foldable f, Applicative m)+ => f o+ -> StreamT o m ()+yieldS x =+ makeSource (F.toList x) go+ where+ go [] = pure (Done ())+ go (y:ys) = pure (Yield ys y)+{-# INLINE yieldS #-}++mapS :: Applicative m+ => (a -> b)+ -> StreamT a m r+ -> StreamT b m r+mapS f =+ makeTransformer' go+ where+ go s g = fmap (mapStep f) (g s)+{-# INLINE mapS #-}++foldlS :: Monad m+ => (accum -> a -> accum)+ -> accum+ -> StreamT a m r+ -> m accum+foldlS f accum0 =+ makeSink go+ where+ go s0 g =+ loop accum0 s0+ where+ loop accum s = do+ s <- g s+ case s of+ Done _ -> pure accum+ Yield s' a ->+ let !accum' = f accum a+ in loop accum' s'+ Skip s' -> loop accum s'+{-# INLINE foldlS #-}++sumS :: (Monad m, Num a)+ => StreamT a m r+ -> m a+sumS = foldlS (+) 0+{-# INLINE sumS #-}++sinkListS :: Monad m+ => StreamT i m r+ -> m [i]+sinkListS =+ fmap ($ []) . foldlS go id+ where+ go front x = front . (x:)+{-# INLINE sinkListS #-}++mapM_S :: Monad m+ => (i -> m a)+ -> StreamT i m r+ -> m ()+mapM_S f =+ makeSink go+ where+ go s0 g =+ let loop s = do+ step <- g s+ case step of+ Done _ -> pure ()+ Yield s' i -> f i *> loop s'+ Skip s' -> loop s'+ in loop s0+{-# INLINE mapM_S #-}++readFileS :: (Catch.MonadMask m, MonadIO m)+ => FilePath+ -> StreamT S.ByteString m ()+readFileS fp = makeSourceWith+ (Catch.bracket+ (liftIO (FR.openFile fp))+ (liftIO . FR.closeFile))+ go+ where+ go h =+ liftIO $ fmap toStep $ FR.readChunk h+ where+ toStep bs+ | S.null bs = Done ()+ | otherwise = Yield h bs+{-# INLINE readFileS #-}++writeFileS :: (Catch.MonadMask m, MonadIO m)+ => FilePath+ -> StreamT S.ByteString m ()+ -> m ()+writeFileS fp stream = Catch.bracket+ (liftIO $ IO.openBinaryFile fp IO.WriteMode)+ (liftIO . IO.hClose)+ (\h -> mapM_S (liftIO . S.hPut h) stream)+{-# INLINE writeFileS #-}++linesAsciiS :: Monad m+ => (accum -> StreamT S.ByteString m () -> StreamT o m accum)+ -> accum+ -> StreamT S.ByteString m r+ -> StreamT o m accum+linesAsciiS inner accum0 =+ makeTransformer accum0 go+ where+ go accum s f = do+ step <- f s+ case step of+ Done _ -> pure (Done accum)+ Skip s' -> pure (Skip (accum, s'))+ Yield s' bs+ | S.null bs -> pure (Skip (accum, s'))+ | otherwise -> error "FIXME maybe this isn't possible after all"+{-# INLINE linesAsciiS #-}
+ src/Stream/Core.hs view
@@ -0,0 +1,83 @@+-- FIXME move to stream-core library+{-# LANGUAGE RankNTypes #-}+module Stream.Core+ ( -- * Types+ StreamT+ , Step (..)+ -- * Helpers+ , mapStep+ , runStreamT+ -- * Smart constructros+ -- ** Source+ , makeSource+ , makeSourceWith+ -- ** Transformer+ , makeTransformer+ , makeTransformer'+ -- ** Sink+ , makeSink+ ) where++import Stream.Core.Internal++mapStep :: (i -> o) -> Step s i r -> Step s o r+mapStep _ (Done r) = Done r+mapStep f (Yield s i) = Yield s (f i)+mapStep _ (Skip s) = Skip s+{-# INLINE mapStep #-}++runStreamT :: Monad m => StreamT o m r -> m r+runStreamT =+ makeSink go+ where+ go s0 f =+ loop s0+ where+ loop s = do+ step <- f s+ case step of+ Done r -> pure r+ Yield s' _ -> loop s'+ Skip s' -> loop s'+{-# INLINE runStreamT #-}++makeSource :: state+ -> (state -> m (Step state o r))+ -> StreamT o m r+makeSource state f = StreamT f ($ state)+{-# INLINE makeSource #-}++makeSourceWith :: (forall b. (state -> m b) -> m b)+ -> (state -> m (Step state o r))+ -> StreamT o m r+makeSourceWith withState f = StreamT f withState+{-# INLINE makeSourceWith #-}++makeTransformer :: myState+ -> (forall upState.+ myState+ -> upState+ -> (upState -> m (Step upState i upR))+ -> m (Step (myState, upState) o myR))+ -> StreamT i m upR+ -> StreamT o m myR+makeTransformer myState f (StreamT g withUpState) =+ StreamT (\(myState, upState) -> f myState upState g) withState+ where+ withState inner = withUpState $ \upState -> inner (myState, upState)++makeTransformer' :: (forall upState.+ upState+ -> (upState -> m (Step upState i upR))+ -> m (Step upState o myR))+ -> StreamT i m upR+ -> StreamT o m myR+makeTransformer' f (StreamT g withState) =+ StreamT (\s -> f s g) withState+{-# INLINE makeTransformer' #-}++makeSink :: (forall state. state -> (state -> m (Step state i upR)) -> m myR)+ -> StreamT i m upR+ -> m myR+makeSink f (StreamT g withState) = withState $ \state -> f state g+{-# INLINE makeSink #-}
+ src/Stream/Core/Internal.hs view
@@ -0,0 +1,13 @@+-- FIXME move to stream-core library+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE GADTs #-}+module Stream.Core.Internal where++data Step s o r+ = Done r+ | Yield s o+ | Skip s++data StreamT o m r = forall s. StreamT+ (s -> m (Step s o r))+ (forall b. (s -> m b) -> m b)
+ stream.cabal view
@@ -0,0 +1,57 @@+name: stream+version: 0.1.0.0+synopsis: Initial project template from stack+description: Please see README.md+homepage: https://github.com/githubuser/stream#readme+license: MIT+license-file: LICENSE+author: Michael Snoyman+maintainer: michael@fpcomplete.com+copyright: 2016 FP Complete+category: Data+build-type: Simple+extra-source-files: README.md ChangeLog.md+cabal-version: >=1.10++library+ hs-source-dirs: src+ exposed-modules: Stream+ Stream.Core+ Stream.Core.Internal+ build-depends: base >= 4.7 && < 5+ , bytestring+ , exceptions+ , mtl+ , streaming-commons+ , transformers+ default-language: Haskell2010++test-suite stream-test+ type: exitcode-stdio-1.0+ hs-source-dirs: test+ main-is: Spec.hs+ build-depends: base+ , bytestring+ , hspec+ , stream+ , temporary+ ghc-options: -threaded -rtsopts -with-rtsopts=-N+ default-language: Haskell2010++benchmark stream-bench+ type: exitcode-stdio-1.0+ hs-source-dirs: bench+ main-is: stream-bench.hs+ build-depends: base+ , bytestring+ , criterion+ , ghc-prim+ , stream+ , temporary+ , vector+ ghc-options: -threaded -rtsopts -with-rtsopts=-N -O2+ default-language: Haskell2010++source-repository head+ type: git+ location: https://github.com/githubuser/stream
+ test/Spec.hs view
@@ -0,0 +1,31 @@+import Test.Hspec+import Test.Hspec.QuickCheck+import Stream+import qualified Data.ByteString as S+import qualified Data.ByteString.Char8 as S8+import System.IO.Temp+import System.IO (hClose)++main :: IO ()+main = hspec $ do+ it "enum/map/sum sanity" $+ runIdentity (sumS $ mapS (+ 1) $ enumFromToS 1 (1000 :: Int))+ `shouldBe` sum (map (+ 1) [1..1000])+ prop "file copy" $ \octets ->+ withSystemTempFile "src" $ \srcFP srcH ->+ withSystemTempFile "dst" $ \dstFP dstH -> do+ let bsOrig = S.pack octets+ S.hPut srcH bsOrig+ hClose srcH+ hClose dstH+ writeFileS dstFP $ readFileS srcFP+ actual <- S.readFile dstFP+ actual `shouldBe` bsOrig+ prop "lines" $ \octetss ->+ let ls = map S.pack octetss+ bs = S8.unlines ls+ src = yieldS [bs]+ sink = linesAsciiS (const go) ()+ go = error "go"+ res = runIdentity $ sinkListS $ sink src+ in res `shouldBe` S8.lines bs