hw-kafka-conduit 1.1.1 → 1.1.2
raw patch · 6 files changed
+94/−74 lines, 6 filesdep +QuickCheckdep +extra
Dependencies added: QuickCheck, extra
Files
- hw-kafka-conduit.cabal +48/−47
- src/Kafka/Conduit/Combinators.hs +43/−0
- src/Kafka/Conduit/Sink.hs +1/−1
- src/Kafka/Conduit/Source.hs +1/−1
- src/Kafka/Conduit/Utils.hs +0/−23
- test/Spec.hs +1/−2
hw-kafka-conduit.cabal view
@@ -1,5 +1,5 @@ name: hw-kafka-conduit-version: 1.1.1+version: 1.1.2 synopsis: Conduit bindings for kafka-client homepage: https://github.com/haskell-works/hw-kafka-client-conduit bug-reports: https://github.com/haskell-works/hw-kafka-client-conduit/issues@@ -19,53 +19,54 @@ location: https://github.com/haskell-works/hw-kafka-conduit executable kafka-conduit-example- main-is: Main.hs- hs-source-dirs: example- default-language: Haskell2010- build-depends: base >=4.6 && < 5- , bifunctors- , bytestring- , conduit- , containers- , resourcet- , hw-kafka-conduit+ main-is: Main.hs+ hs-source-dirs: example+ default-language: Haskell2010+ build-depends: base >=4.6 && < 5+ , bifunctors+ , bytestring+ , conduit+ , containers+ , resourcet+ , hw-kafka-conduit library- hs-source-dirs: src- ghc-options: -Wall- exposed-modules: Kafka.Conduit- , Kafka.Conduit.Source- , Kafka.Conduit.Sink- , Kafka.Conduit.Utils- default-language: Haskell2010- build-depends: base >= 4.7 && < 5- , bifunctors- , bytestring- , containers- , conduit- , conduit-extra- , exceptions- , hw-kafka-client- , mtl- , resourcet- , transformers-+ hs-source-dirs: src+ ghc-options: -Wall+ exposed-modules: Kafka.Conduit+ , Kafka.Conduit.Combinators+ , Kafka.Conduit.Source+ , Kafka.Conduit.Sink+ default-language: Haskell2010+ build-depends: base >= 4.7 && < 5+ , bifunctors+ , bytestring+ , containers+ , conduit+ , conduit-extra+ , exceptions+ , hw-kafka-client+ , mtl+ , resourcet+ , transformers test-suite kafka-client-conduit-test- type: exitcode-stdio-1.0- hs-source-dirs: test- main-is: Spec.hs- ghc-options: -Wall -threaded -rtsopts -with-rtsopts=-N- default-language: Haskell2010- build-depends: base- , hw-kafka-conduit- , bifunctors- , bytestring- , containers- , conduit- , conduit-extra- , hspec- , hw-kafka-client- , mtl- , resourcet- , transformers+ type: exitcode-stdio-1.0+ hs-source-dirs: test+ main-is: Spec.hs+ ghc-options: -Wall -threaded -rtsopts -with-rtsopts=-N+ default-language: Haskell2010+ build-depends: base+ , hw-kafka-conduit+ , bifunctors+ , bytestring+ , containers+ , conduit+ , conduit-extra+ , extra+ , hspec+ , hw-kafka-client+ , mtl+ , resourcet+ , transformers+ , QuickCheck
+ src/Kafka/Conduit/Combinators.hs view
@@ -0,0 +1,43 @@+module Kafka.Conduit.Combinators+ ( batchByOrFlush+ , foldYield+ , throwLeft+ , throwLeftSatisfy+ ) where++import Control.Exception+import Control.Monad+import Control.Monad.Catch+import Data.Conduit++-- | Throws the left part of a value in a 'MonadThrow' context+throwLeft :: (MonadThrow m, Exception e) => Conduit (Either e i) m i+throwLeft = awaitForever (either throwM yield)++-- | Throws the left part of a value in a 'MonadThrow' context if the value+-- satisfies the predicate+throwLeftSatisfy :: (MonadThrow m, Exception e) => (e -> Bool) -> Conduit (Either e i) m (Either e i)+throwLeftSatisfy p = awaitForever awaitHandle+ where awaitHandle (Left e) | p e = throwM e+ awaitHandle v = yield v++-- | Create a conduit that folds with the function f over its input i with its+-- internal state s and emits outputs [o], then finally emits outputs [o] from+-- the function g applied to the final state s.+foldYield :: Monad m => (i -> s -> (s, [o])) -> (s -> [o]) -> s -> Conduit i m o+foldYield f g s = do+ mi <- await+ case mi of+ Just i -> do+ let (s', os) = f i s+ forM_ os yield+ foldYield f g s'+ Nothing -> forM_ (g s) yield++batchByOrFlush :: Monad m => Int -> Conduit (Maybe a) m [a]+batchByOrFlush n = foldYield folder finish (0 :: Int, [])+ where+ folder Nothing (_, xs) = ((0 , []), [reverse xs ])+ folder (Just a) (i, xs) | (i + 1) >= n = ((0 , []), [reverse (a:xs)])+ folder (Just a) (i, xs) = ((i + 1, a:xs), [])+ finish (_, xs) = [reverse xs]
src/Kafka/Conduit/Sink.hs view
@@ -8,7 +8,7 @@ import Control.Monad.Trans.Resource import Data.Conduit import Kafka.Producer as X-import Kafka.Conduit.Utils as X+import Kafka.Conduit.Combinators as X -- | Creates a Sink for a given `KafkaProducer`. -- The producer will be closed when the Sink is closed.
src/Kafka/Conduit/Source.hs view
@@ -23,7 +23,7 @@ import Data.Conduit import qualified Data.Conduit.List as L import Kafka.Consumer as X-import Kafka.Conduit.Utils as X+import Kafka.Conduit.Combinators as X -- | Create a `Source` for a given `KafkaConsumer`. -- The consumer will NOT be closed automatically when the `Source` is closed.
− src/Kafka/Conduit/Utils.hs
@@ -1,23 +0,0 @@-module Kafka.Conduit.Utils where--import Control.Exception-import Control.Monad-import Control.Monad.Catch-import Data.Conduit---- | Throws the left part of a value in a 'MonadThrow' context-throwLeft :: (MonadThrow m, Exception e) => Conduit (Either e i) m i-throwLeft = awaitForever (either throwM yield)---- | Create a conduit that folds with the function f over its input i with its--- internal state s and emits outputs [o], then finally emits outputs [o] from--- the function g applied to the final state s.-foldYield :: Monad m => (i -> s -> (s, [o])) -> (s -> [o]) -> s -> Conduit i m o-foldYield f g s = do- mi <- await- case mi of- Just i -> do- let (s', os) = f i s- forM_ os yield- foldYield f g s'- Nothing -> forM_ (g s) yield
test/Spec.hs view
@@ -1,2 +1,1 @@-main :: IO ()-main = putStrLn "Test suite not yet implemented"+{-# OPTIONS_GHC -F -pgmF hspec-discover #-}