diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -0,0 +1,22 @@ +Copyright (c) 2019, Metrix.AI + +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. diff --git a/Setup.hs b/Setup.hs new file mode 100644 --- /dev/null +++ b/Setup.hs @@ -0,0 +1,2 @@ +import Distribution.Simple +main = defaultMain diff --git a/library/Potoki/Conduit/MonadIO.hs b/library/Potoki/Conduit/MonadIO.hs new file mode 100644 --- /dev/null +++ b/library/Potoki/Conduit/MonadIO.hs @@ -0,0 +1,37 @@ +{-| Integration in "MonadIO". -} +module Potoki.Conduit.MonadIO +( + consumeConduit, +) +where + +import Potoki.Conduit.Prelude +import Conduit +import qualified Potoki.Core.Produce as Produce +import qualified Potoki.Core.Transform as Transform +import qualified Potoki.Core.Consume as Consume +import qualified Potoki.Core.Fetch as Fetch +import qualified Potoki.Core.IO as IO + + +{-| Given a Conduit source over a "MonadIO" monad and a Potoki consumer + execute the whole pipeline in the base monad of the source. -} +consumeConduit :: MonadIO m => ConduitT () o m () -> Consume.Consume o r -> m r +consumeConduit conduit consume = do + elementChan <- liftIO $ newTBMChanIO 100 + resultVar <- liftIO $ newEmptyMVar + liftIO $ fork $ do + result <- IO.produceAndConsume (Produce.tbmChan elementChan) consume + putMVar resultVar result + feedConduitToTBMChan conduit elementChan + liftIO $ takeMVar resultVar + +feedConduitToMVar :: MonadIO m => ConduitT () o m () -> MVar (Maybe o) -> m () +feedConduitToMVar conduit mvar = do + runConduit $ conduit .| mapM_C (liftIO . putMVar mvar . Just) + liftIO (putMVar mvar Nothing) + +feedConduitToTBMChan :: MonadIO m => ConduitT () o m () -> TBMChan o -> m () +feedConduitToTBMChan conduit chan = do + runConduit $ conduit .| mapM_C (liftIO . atomically . writeTBMChan chan) + liftIO $ atomically $ closeTBMChan chan diff --git a/library/Potoki/Conduit/Prelude.hs b/library/Potoki/Conduit/Prelude.hs new file mode 100644 --- /dev/null +++ b/library/Potoki/Conduit/Prelude.hs @@ -0,0 +1,93 @@ +module Potoki.Conduit.Prelude +( + module Exports, +) +where + +-- base +------------------------- +import Control.Applicative as Exports +import Control.Arrow as Exports +import Control.Category as Exports +import Control.Concurrent as Exports hiding (forkFinally) +import Control.Exception as Exports +import Control.Monad as Exports hiding (mapM_, sequence_, forM_, msum, mapM, sequence, forM) +import Control.Monad.IO.Class as Exports +import Control.Monad.Fix as Exports hiding (fix) +import Control.Monad.ST as Exports +import Data.Bits as Exports +import Data.Bool as Exports +import Data.Char as Exports +import Data.Coerce as Exports +import Data.Complex as Exports +import Data.Data as Exports +import Data.Dynamic as Exports +import Data.Either as Exports +import Data.Fixed as Exports +import Data.Foldable as Exports hiding (toList) +import Data.Function as Exports hiding (id, (.)) +import Data.Functor as Exports +import Data.Functor.Identity as Exports +import Data.Int as Exports +import Data.IORef as Exports +import Data.Ix as Exports +import Data.List as Exports hiding (sortOn, isSubsequenceOf, uncons, concat, foldr, foldl1, maximum, minimum, product, sum, all, and, any, concatMap, elem, foldl, foldr1, notElem, or, find, maximumBy, minimumBy, mapAccumL, mapAccumR, foldl') +import Data.Maybe as Exports +import Data.Monoid as Exports hiding (Last(..), First(..), (<>)) +import Data.Ord as Exports +import Data.Proxy as Exports +import Data.Ratio as Exports +import Data.Semigroup as Exports +import Data.STRef as Exports +import Data.String as Exports +import Data.Traversable as Exports +import Data.Tuple as Exports +import Data.Unique as Exports +import Data.Version as Exports +import Data.Word as Exports +import Debug.Trace as Exports +import Foreign.ForeignPtr as Exports +import Foreign.Ptr as Exports +import Foreign.StablePtr as Exports +import Foreign.Storable as Exports hiding (sizeOf, alignment) +import GHC.Conc as Exports hiding (withMVar, threadWaitWriteSTM, threadWaitWrite, threadWaitReadSTM, threadWaitRead) +import GHC.Exts as Exports (lazy, inline, sortWith, groupWith, IsList(..)) +import GHC.Generics as Exports (Generic) +import GHC.IO.Exception as Exports +import Numeric as Exports +import Prelude as Exports hiding (concat, foldr, mapM_, sequence_, foldl1, maximum, minimum, product, sum, all, and, any, concatMap, elem, foldl, foldr1, notElem, or, mapM, sequence, id, (.)) +import System.Environment as Exports +import System.Exit as Exports +import System.IO as Exports +import System.IO.Error as Exports +import System.IO.Unsafe as Exports +import System.Mem as Exports +import System.Mem.StableName as Exports +import System.Timeout as Exports +import Text.ParserCombinators.ReadP as Exports (ReadP, ReadS, readP_to_S, readS_to_P) +import Text.ParserCombinators.ReadPrec as Exports (ReadPrec, readPrec_to_P, readP_to_Prec, readPrec_to_S, readS_to_Prec) +import Text.Printf as Exports (printf, hPrintf) +import Text.Read as Exports (Read(..), readMaybe, readEither) +import Unsafe.Coerce as Exports + +-- bytestring +------------------------- +import Data.ByteString as Exports (ByteString) + +-- text +------------------------- +import Data.Text as Exports (Text) + +-- profunctors +------------------------- +import Data.Profunctor.Unsafe as Exports +import Data.Profunctor.Choice as Exports +import Data.Profunctor.Strong as Exports + +-- slave-thread +------------------------- +import SlaveThread as Exports + +-- stm-chans +------------------------- +import Control.Concurrent.STM.TBMChan as Exports diff --git a/potoki-conduit.cabal b/potoki-conduit.cabal new file mode 100644 --- /dev/null +++ b/potoki-conduit.cabal @@ -0,0 +1,50 @@ +name: potoki-conduit +version: 0.1 +synopsis: Integration of "potoki" and "conduit" +category: Potoki, Streaming, Conduit +homepage: https://github.com/metrix-ai/potoki-conduit +bug-reports: https://github.com/metrix-ai/potoki-conduit/issues +author: Nikita Volkov +maintainer: Metrix.AI Tech Team +copyright: (c) 2019, Metrix.AI +license: MIT +license-file: LICENSE +build-type: Simple +cabal-version: >=1.10 + +library + hs-source-dirs: library + default-extensions: Arrows, BangPatterns, ConstraintKinds, DataKinds, DefaultSignatures, DeriveDataTypeable, DeriveFoldable, DeriveFunctor, DeriveGeneric, DeriveTraversable, EmptyDataDecls, FlexibleContexts, FlexibleInstances, FunctionalDependencies, GADTs, GeneralizedNewtypeDeriving, LambdaCase, LiberalTypeSynonyms, MagicHash, MultiParamTypeClasses, MultiWayIf, NoImplicitPrelude, NoMonomorphismRestriction, OverloadedStrings, PatternGuards, ParallelListComp, QuasiQuotes, RankNTypes, RecordWildCards, ScopedTypeVariables, StandaloneDeriving, TemplateHaskell, TupleSections, TypeFamilies, TypeOperators, UnboxedTuples + default-language: Haskell2010 + exposed-modules: + Potoki.Conduit.MonadIO + other-modules: + Potoki.Conduit.Prelude + build-depends: + acquire >=0.2 && <0.3, + base >=4.7 && <5, + base-prelude >=1 && <2, + bytestring >=0.10 && <0.11, + conduit >=1.3 && <1.4, + potoki-core >=2.3.4 && <2.4, + profunctors >=5.2 && <6, + slave-thread >=1.0.3 && <2, + stm-chans >=3 && <3.1, + text >=1 && <2 + +test-suite test + type: exitcode-stdio-1.0 + hs-source-dirs: test + default-extensions: Arrows, BangPatterns, ConstraintKinds, DataKinds, DefaultSignatures, DeriveDataTypeable, DeriveFoldable, DeriveFunctor, DeriveGeneric, DeriveTraversable, EmptyDataDecls, FlexibleContexts, FlexibleInstances, FunctionalDependencies, GADTs, GeneralizedNewtypeDeriving, LambdaCase, LiberalTypeSynonyms, MagicHash, MultiParamTypeClasses, MultiWayIf, NoImplicitPrelude, NoMonomorphismRestriction, OverloadedStrings, PatternGuards, ParallelListComp, QuasiQuotes, RankNTypes, RecordWildCards, ScopedTypeVariables, StandaloneDeriving, TemplateHaskell, TupleSections, TypeFamilies, TypeOperators, UnboxedTuples + default-language: Haskell2010 + main-is: Main.hs + build-depends: + conduit, + potoki >=2.1.4 && <2.2, + potoki-conduit, + QuickCheck >=2.8.1 && <3, + quickcheck-instances >=0.3.11 && <0.4, + rerebase <2, + tasty >=0.12 && <2, + tasty-hunit >=0.9 && <0.11, + tasty-quickcheck >=0.9 && <0.11 diff --git a/test/Main.hs b/test/Main.hs new file mode 100644 --- /dev/null +++ b/test/Main.hs @@ -0,0 +1,24 @@ +module Main where + +import Prelude hiding (choose) +import Test.QuickCheck.Instances +import Test.Tasty +import Test.Tasty.Runners +import Test.Tasty.HUnit +import Test.Tasty.QuickCheck +import qualified Test.QuickCheck as QuickCheck +import qualified Test.QuickCheck.Property as QuickCheck +import qualified Potoki.Conduit.MonadIO as MonadIO +import qualified Potoki.Consume as Consume +import qualified Potoki.IO as IO +import qualified Conduit + + +main = + defaultMain $ + testGroup "All" $ + [ + testProperty "List restoration" $ \ (list :: [Int]) -> unsafePerformIO $ do + restoredList <- MonadIO.consumeConduit (Conduit.yieldMany list) Consume.list + return (list === restoredList) + ]