hw-kafka-conduit 1.1.4 → 2.0.0

raw patch · 4 files changed

+34/−39 lines, 4 files

Files

• hw-kafka-conduit.cabal +2/−3
• src/Kafka/Conduit.hs +0/−6
• src/Kafka/Conduit/Sink.hs +13/−10
• src/Kafka/Conduit/Source.hs +19/−20

@@ -1,5 +1,5 @@ name:                hw-kafka-conduit-version:             1.1.4+version:             2.0.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@@ -42,8 +42,7 @@ library   hs-source-dirs:       src   ghc-options:          -Wall-  exposed-modules:      Kafka.Conduit-                      , Kafka.Conduit.Combinators+  exposed-modules:      Kafka.Conduit.Combinators                       , Kafka.Conduit.Source                       , Kafka.Conduit.Sink   default-language:     Haskell2010

@@ -1,6 +0,0 @@-module Kafka.Conduit-( module X-) where--import Kafka.Conduit.Source as X-import Kafka.Conduit.Sink as X

@@ -4,14 +4,17 @@ , commitOffsetsSink, flushThenCommitSink ) where -import Control.Monad.IO.Class-import Control.Monad (void)-import Control.Monad.Trans.Resource-import Data.Conduit-import Kafka.Consumer as X-import Kafka.Producer as X-import Kafka.Conduit.Combinators as X+import           Control.Monad                (void)+import           Control.Monad.IO.Class+import           Control.Monad.Trans.Resource+import           Data.Conduit+import           Kafka.Consumer +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@@ -27,7 +30,7 @@         Just msg -> do           res <- produceMessage p' msg           case res of-            Nothing -> runHandler p'+            Nothing  -> runHandler p'             Just err -> return (Just err)  -- | Creates a Sink for a given `KafkaProducer`.@@ -44,7 +47,7 @@         Just msg -> do           res <- produceMessage prod msg           case res of-            Nothing -> go+            Nothing  -> go             Just err -> return (Just err)  -- | Creates a batching Sink for a given `KafkaProducer`.@@ -77,7 +80,7 @@   where     mkProducer = newProducer props -    clProducer (Left _) = return ()+    clProducer (Left _)     = return ()     clProducer (Right prod) = void $ closeProducer prod      runHandler (Left err) = return (Just err)

@@ -1,4 +1,3 @@-{-# LANGUAGE TupleSections#-} module Kafka.Conduit.Source ( -- ** Source   kafkaSource, kafkaSourceNoClose, kafkaSourceAutoClose@@ -14,16 +13,16 @@ , module X ) where -import Data.Bifunctor-import Data.Bitraversable-import Control.Monad.IO.Class-import Control.Monad (void)-import Control.Monad.Trans.Resource-import qualified Data.ByteString as BS-import Data.Conduit-import qualified Data.Conduit.List as L-import Kafka.Consumer as X-import Kafka.Conduit.Combinators as X+import           Control.Monad                (void)+import           Control.Monad.IO.Class+import           Control.Monad.Trans.Resource+import           Data.Bifunctor+import           Data.Bitraversable+import qualified Data.ByteString              as BS+import           Data.Conduit+import qualified Data.Conduit.List            as L+import           Kafka.Conduit.Combinators    as X+import           Kafka.Consumer               as X  -- | Create a `Source` for a given `KafkaConsumer`. -- The consumer will NOT be closed automatically when the `Source` is closed.@@ -38,7 +37,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 >> go+        _        -> yield msg >> go  -- | Create a `Source` for a given `KafkaConsumer`. -- The consumer will be closed automatically when the `Source` is closed.@@ -56,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`. --@@ -73,7 +72,7 @@   where       mkConsumer = newConsumer props sub -      clConsumer (Left _) = return ()+      clConsumer (Left _)  = return ()       clConsumer (Right c) = void $ closeConsumer c        runHandler (Left err) = void $ yield (Left err)@@ -82,7 +81,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 (Right c)+          _        -> yield msg >> runHandler (Right c)  ------------------------------- Utitlity functions @@ -223,10 +222,10 @@ -- or is unsafe to ignore. isFatal :: KafkaError -> Bool isFatal e = case e of-  KafkaUnknownConfigurationKey _   -> True-  KafkaInvalidConfigurationValue _ -> True-  KakfaBadConfiguration            -> True-  KafkaBadSpecification _          -> True+  KafkaUnknownConfigurationKey _                              -> True+  KafkaInvalidConfigurationValue _                            -> True+  KakfaBadConfiguration                                       -> True+  KafkaBadSpecification _                                     -> True    -- More of them? Less of them?   KafkaResponseError RdKafkaRespErrDestroy                    -> True@@ -244,4 +243,4 @@   KafkaResponseError RdKafkaRespErrIllegalSaslState           -> True   KafkaResponseError RdKafkaRespErrUnsupportedVersion         -> True -  _ -> False+  _                                                           -> False