packages feed

hw-kafka-conduit 2.5.0 → 2.6.0

raw patch · 4 files changed

+49/−49 lines, 4 files

Files

hw-kafka-conduit.cabal view
@@ -1,5 +1,5 @@ name:                hw-kafka-conduit-version:             2.5.0+version:             2.6.0 synopsis:            Conduit bindings for hw-kafka-client homepage:            https://github.com/haskell-works/hw-kafka-conduit bug-reports:         https://github.com/haskell-works/hw-kafka-conduit/issues@@ -12,7 +12,7 @@ build-type:          Simple -- extra-source-files: cabal-version:       >=1.10-description:         Conduit bindings for hw-kafka-client+description:         Conduit bindings for hw-kafka-client.  flag examples   default: False
src/Kafka/Conduit/Combinators.hs view
@@ -7,18 +7,18 @@   , throwLeftSatisfy   ) where -import           Control.Monad       (forM_)-import           Control.Monad.Catch (Exception, MonadThrow, throwM)-import           Data.Conduit        (Conduit, await, awaitForever, yield)-import           Kafka.Types         (BatchSize (..))+import Control.Monad       (forM_)+import Control.Monad.Catch (Exception, MonadThrow, throwM)+import Data.Conduit        (ConduitT, await, awaitForever, yield)+import Kafka.Types         (BatchSize (..))  -- | Throws the left part of a value in a 'MonadThrow' context-throwLeft :: (MonadThrow m, Exception e) => Conduit (Either e i) m i+throwLeft :: (MonadThrow m, Exception e) => ConduitT (Either e i) i m () 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 :: (MonadThrow m, Exception e) => (e -> Bool) -> ConduitT (Either e i) (Either e i) m () throwLeftSatisfy p = awaitForever awaitHandle   where awaitHandle (Left e) | p e  = throwM e         awaitHandle v        = yield v@@ -26,7 +26,7 @@ -- | 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 :: Monad m => (i -> s -> (s, [o])) -> (s -> [o]) -> s -> ConduitT i o m () foldYield f g s = do   mi <- await   case mi of@@ -36,15 +36,15 @@       foldYield f g s'     Nothing -> forM_ (g s) yield -batchByOrFlush :: Monad m => BatchSize -> Conduit (Maybe a) m [a]+batchByOrFlush :: Monad m => BatchSize -> ConduitT (Maybe a) [a] m () batchByOrFlush (BatchSize n) = foldYield folder finish (0 :: Int, [])   where-    folder Nothing  (_, xs)                 = ((0    ,   []), [reverse    xs ])+    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),               [])+    folder (Just a) (i, xs) = ((i + 1, a:xs),               [])     finish (_, xs) = [reverse xs] -batchByOrFlushEither :: Monad m => BatchSize -> Conduit (Either e a) m [a]+batchByOrFlushEither :: Monad m => BatchSize -> ConduitT (Either e a) [a] m () batchByOrFlushEither (BatchSize n) = foldYield folder finish (0 :: Int, [])   where     folder (Left _)  (_, xs) = ((0    ,    []), [reverse    xs ])
src/Kafka/Conduit/Sink.hs view
@@ -11,16 +11,16 @@ import qualified Data.Conduit.List            as L import           Kafka.Consumer -import           Kafka.Conduit.Combinators    as X-import           Kafka.Consumer               as X (KafkaConsumer)-import           Kafka.Producer               as X+import Kafka.Conduit.Combinators as X+import Kafka.Consumer            as X (KafkaConsumer)+import Kafka.Producer            as X   -- | Creates a Sink for a given `KafkaProducer`. -- The producer will be closed when the Sink is closed. kafkaSinkAutoClose :: MonadResource m                    => KafkaProducer-                   -> Sink ProducerRecord m (Maybe KafkaError)+                   -> ConduitT ProducerRecord Void m (Maybe KafkaError) kafkaSinkAutoClose prod =   bracketP (return prod) (void . closeProducer) runHandler   where@@ -38,7 +38,7 @@ -- The producer will NOT be closed automatically. kafkaSinkNoClose :: MonadIO m                  => KafkaProducer-                 -> Sink ProducerRecord m (Maybe KafkaError)+                 -> ConduitT ProducerRecord Void m (Maybe KafkaError) kafkaSinkNoClose prod = go   where     go = do@@ -55,7 +55,7 @@ -- The producer will NOT be closed automatically. kafkaBatchSinkNoClose :: MonadIO m                  => KafkaProducer-                 -> Sink [ProducerRecord] m [(ProducerRecord, KafkaError)]+                 -> ConduitT [ProducerRecord] Void m [(ProducerRecord, KafkaError)] kafkaBatchSinkNoClose prod = go   where     go = do@@ -75,7 +75,7 @@ -- 'kafkaSinkAutoClose' or 'kafkaSinkNoClose' can be used. kafkaSink :: MonadResource m           => ProducerProperties-          -> Sink ProducerRecord m (Maybe KafkaError)+          -> ConduitT ProducerRecord Void m (Maybe KafkaError) kafkaSink props =   bracketP mkProducer clProducer runHandler   where@@ -99,7 +99,7 @@ -- -- > mapMC (\_ -> commitAllOffsets OffsetCommit consumer) {-# DEPRECATED commitOffsetsSink "Conceptually wrong thing to do. Does not require library support. Consider calling 'commitAllOffsets' when appropriate." #-}-commitOffsetsSink :: MonadIO m => KafkaConsumer -> Sink i m ()+commitOffsetsSink :: MonadIO m => KafkaConsumer -> ConduitT i Void m () commitOffsetsSink = flip commitOffsetsSink' (const $ pure ())  -- | Ignores incoming messages and commits offsets. Commit errors are handled with 'handleError' effect.@@ -108,7 +108,7 @@ -- -- > mapMC (\_ -> commitAllOffsets OffsetCommit consumer >>= handleError) {-# DEPRECATED commitOffsetsSink' "Conceptually wrong thing to do. Does not require library support. Consider calling 'commitAllOffsets' when appropriate." #-}-commitOffsetsSink':: MonadIO m => KafkaConsumer -> (KafkaError -> m ()) -> Sink i m ()+commitOffsetsSink':: MonadIO m => KafkaConsumer -> (KafkaError -> m ()) -> ConduitT i Void m () commitOffsetsSink' consumer handleError = L.mapM_ $ \_ -> do   res <- commitAllOffsets OffsetCommit consumer   case res of@@ -122,7 +122,7 @@ -- -- > mapMC (\_ -> flushProducer producer >>= commitAllOffsets OffsetCommit consumer) {-# DEPRECATED flushThenCommitSink "Conceptually wrong thing to do. Does not require library support. Consider calling 'flushProducer >>= commitAllOffsets' when appropriate." #-}-flushThenCommitSink :: MonadIO m => KafkaConsumer -> KafkaProducer -> Sink i m ()+flushThenCommitSink :: MonadIO m => KafkaConsumer -> KafkaProducer -> ConduitT i Void m () flushThenCommitSink consumer producer = flushThenCommitSink' consumer producer (const $ pure ())  -- | Ignores incoming messages and commits offsets, but makes sure that 'producer' has an empty outgoing queue.@@ -132,7 +132,7 @@ -- -- > mapMC (\_ -> flushProducer producer >>= commitAllOffsets OffsetCommit consumer >>= handleError) {-# DEPRECATED flushThenCommitSink' "Conceptually wrong thing to do. Does not require library support. Consider calling 'flushProducer >>= commitAllOffsets' when appropriate." #-}-flushThenCommitSink' :: MonadIO m => KafkaConsumer -> KafkaProducer -> (KafkaError -> m ()) -> Sink i m ()+flushThenCommitSink' :: MonadIO m => KafkaConsumer -> KafkaProducer -> (KafkaError -> m ()) -> ConduitT i Void m () flushThenCommitSink' consumer producer handleError = L.mapM_ $ \_ -> do   flushProducer producer   res <- commitAllOffsets OffsetCommit consumer
src/Kafka/Conduit/Source.hs view
@@ -29,7 +29,7 @@ kafkaSourceNoClose :: MonadIO m                    => KafkaConsumer                    -> Timeout-                   -> Source m (Either KafkaError (ConsumerRecord (Maybe BS.ByteString) (Maybe BS.ByteString)))+                   -> ConduitT () (Either KafkaError (ConsumerRecord (Maybe BS.ByteString) (Maybe BS.ByteString))) m () kafkaSourceNoClose c t = go   where     go = do@@ -37,14 +37,14 @@       -- stop at some certain cases because it is not goind to be better with time       case msg of         Left err | isFatal err -> void $ yield (Left err)-        _        -> yield msg >> go+        _                      -> yield msg >> go  -- | Create a `Source` for a given `KafkaConsumer`. -- The consumer will be closed automatically when the `Source` is closed. kafkaSourceAutoClose :: MonadResource m                      => KafkaConsumer                      -> Timeout-                     -> Source m (Either KafkaError (ConsumerRecord (Maybe BS.ByteString) (Maybe BS.ByteString)))+                     -> ConduitT () (Either KafkaError (ConsumerRecord (Maybe BS.ByteString) (Maybe BS.ByteString))) m () kafkaSourceAutoClose c ts =   bracketP mkConsumer clConsumer runHandler   where@@ -55,7 +55,7 @@       -- stop at some certain cases because it is not goind to be better with time       case msg of         Left err | isFatal err -> void $ yield (Left err)-        _        -> yield msg >> runHandler c'+        _                      -> yield msg >> runHandler c'  -- | Creates a kafka producer for given properties and returns a `Source`. --@@ -66,7 +66,7 @@             => ConsumerProperties             -> Subscription             -> Timeout              -- ^ Poll timeout-            -> Source m (Either KafkaError (ConsumerRecord (Maybe BS.ByteString) (Maybe BS.ByteString)))+            -> ConduitT () (Either KafkaError (ConsumerRecord (Maybe BS.ByteString) (Maybe BS.ByteString))) m () kafkaSource props sub ts =   bracketP mkConsumer clConsumer runHandler   where@@ -81,100 +81,100 @@         -- stop at some certain cases because it is not goind to be better with time         case msg of           Left err | isFatal err -> void $ yield (Left err)-          _        -> yield msg >> runHandler (Right c)+          _                      -> yield msg >> runHandler (Right c)  ------------------------------- Utitlity functions  -- | Maps over the first element of a value -- -- > mapFirst f = L.map (first f)-mapFirst :: (Bifunctor t, Monad m) => (k -> k') -> Conduit (t k v) m (t k' v)+mapFirst :: (Bifunctor t, Monad m) => (k -> k') -> ConduitT (t k v) (t k' v) m () mapFirst f = L.map (first f) {-# INLINE mapFirst #-}  -- | Maps over a value -- -- > mapValue f = L.map (fmap f)-mapValue :: (Functor t, Monad m) => (v -> v') -> Conduit (t v) m (t v')+mapValue :: (Functor t, Monad m) => (v -> v') -> ConduitT (t v) (t v') m () mapValue f = L.map (fmap f) {-# INLINE mapValue #-}  -- | Bimaps (maps over both the first and the second element) over a value -- -- > bimapValue f g = L.map (bimap f g)-bimapValue :: (Bifunctor t, Monad m) => (k -> k') -> (v -> v') -> Conduit (t k v) m (t k' v')+bimapValue :: (Bifunctor t, Monad m) => (k -> k') -> (v -> v') -> ConduitT (t k v) (t k' v') m () bimapValue f g = L.map (bimap f g) {-# INLINE bimapValue #-}  -- | Sequences the first element of a value -- -- > sequenceValueFirst = L.map sequenceFirst-sequenceValueFirst :: (Bitraversable t, Applicative f, Monad m) => Conduit (t (f k) v) m (f (t k v))-sequenceValueFirst = L.map sequenceFirst+sequenceValueFirst :: (Bitraversable t, Applicative f, Monad m) => ConduitT (t (f k) v) (f (t k v)) m ()+sequenceValueFirst = L.map (bitraverse id pure) {-# INLINE sequenceValueFirst #-}  -- | Sequences the value -- -- > sequenceValue = L.map sequenceA-sequenceValue :: (Traversable t, Applicative f, Monad m) => Conduit (t (f v)) m (f (t v))+sequenceValue :: (Traversable t, Applicative f, Monad m) => ConduitT (t (f v)) (f (t v)) m () sequenceValue = L.map sequenceA {-# INLINE sequenceValue #-}  -- | Sequences both the first and the second element of a value (bisequences the value) -- -- > bisequenceValue = L.map bisequenceA-bisequenceValue :: (Bitraversable t, Applicative f, Monad m) => Conduit (t (f k) (f v)) m (f (t k v))+bisequenceValue :: (Bitraversable t, Applicative f, Monad m) => ConduitT (t (f k) (f v)) (f (t k v)) m () bisequenceValue = L.map bisequenceA {-# INLINE bisequenceValue #-}  -- | Traverses over the first element of a value -- -- > traverseValueFirst f = L.map (traverseFirst f)-traverseValueFirst :: (Bitraversable t, Applicative f, Monad m) => (k -> f k') -> Conduit (t k v) m (f (t k' v))-traverseValueFirst f = L.map (traverseFirst f)+traverseValueFirst :: (Bitraversable t, Applicative f, Monad m) => (k -> f k') -> ConduitT (t k v) (f (t k' v)) m ()+traverseValueFirst f = L.map (bitraverse f pure) {-# INLINE traverseValueFirst #-}  -- | Traverses over the value -- -- > L.map (traverse f)-traverseValue :: (Traversable t, Applicative f, Monad m) => (v -> f v') -> Conduit (t v) m (f (t v'))+traverseValue :: (Traversable t, Applicative f, Monad m) => (v -> f v') -> ConduitT (t v) (f (t v')) m () traverseValue f = L.map (traverse f) {-# INLINE traverseValue #-}  -- | Traverses over both the first and the second elements of a value (bitraverses over a value) -- -- > bitraverseValue f g = L.map (bitraverse f g)-bitraverseValue :: (Bitraversable t, Applicative f, Monad m) => (k -> f k') -> (v -> f v') -> Conduit (t k v) m (f (t k' v'))+bitraverseValue :: (Bitraversable t, Applicative f, Monad m) => (k -> f k') -> (v -> f v') -> ConduitT (t k v) (f (t k' v')) m () bitraverseValue f g = L.map (bitraverse f g) {-# INLINE bitraverseValue #-}  -- | Monadically traverses over the first element of a value -- -- > traverseValueFirstM f = L.mapM (traverseFirstM f)-traverseValueFirstM :: (Bitraversable t, Applicative f, Monad m) => (k -> m (f k')) -> Conduit (t k v) m (f (t k' v))-traverseValueFirstM f = L.mapM (traverseFirstM f)+traverseValueFirstM :: (Bitraversable t, Applicative f, Monad m) => (k -> m (f k')) -> ConduitT (t k v) (f (t k' v)) m ()+traverseValueFirstM f = L.mapM (fmap (bitraverse id pure) . bitraverse f pure) {-# INLINE traverseValueFirstM #-}  -- | Monadically traverses over a value -- -- > traverseValueM f = L.mapM (traverseM f)-traverseValueM :: (Traversable t, Applicative f, Monad m) => (v -> m (f v')) -> Conduit (t v) m (f (t v'))-traverseValueM f = L.mapM (traverseM f)+traverseValueM :: (Traversable t, Applicative f, Monad m) => (v -> m (f v')) -> ConduitT (t v) (f (t v')) m ()+traverseValueM f = L.mapM (fmap sequenceA . traverse f) {-# INLINE traverseValueM #-}  -- | Monadically traverses over both the first and the second elements of a value -- (monadically bitraverses over a value) -- -- > bitraverseValueM f g = L.mapM (bitraverseM f g)-bitraverseValueM :: (Bitraversable t, Applicative f, Monad m) => (k -> m (f k')) -> (v -> m (f v')) -> Conduit (t k v) m (f (t k' v'))-bitraverseValueM f g = L.mapM (bitraverseM f g)+bitraverseValueM :: (Bitraversable t, Applicative f, Monad m) => (k -> m (f k')) -> (v -> m (f v')) -> ConduitT (t k v) (f (t k' v')) m ()+bitraverseValueM f g = L.mapM (fmap bisequenceA . bimapM f g) {-# INLINE bitraverseValueM #-}  --------------------------------------------------------------------------------  -- | Filters out non-fatal errors (see 'isFatal') and only allows fatal errors -- to be propagated downstream.-skipNonFatal :: Monad m => Conduit (Either KafkaError b) m (Either KafkaError b)+skipNonFatal :: Monad m => ConduitT (Either KafkaError b) (Either KafkaError b) m () skipNonFatal = L.filter (either isFatal (const True)) {-# INLINE skipNonFatal #-} @@ -212,7 +212,7 @@ -- "'KafkaResponseError' 'RdKafkaRespErrTimedOut'" and "'KafkaResponseError' 'RdKafkaRespErrPartitionEof'". -- -- This function does not allow filtering out fatal errors.-skipNonFatalExcept :: Monad m => [KafkaError -> Bool] -> Conduit (Either KafkaError b) m (Either KafkaError b)+skipNonFatalExcept :: Monad m => [KafkaError -> Bool] -> ConduitT (Either KafkaError b) (Either KafkaError b) m () skipNonFatalExcept fs =   let fun e = or $ (\f -> f e) <$> (isFatal : fs)    in L.filter (either fun (const True))