packages feed

hw-kafka-conduit 2.0.0 → 2.1.0

raw patch · 4 files changed

+97/−13 lines, 4 files

Files

example/Main.hs view
@@ -1,12 +1,12 @@ {-# LANGUAGE OverloadedStrings #-} module Main where -import Control.Monad (forM_)-import Data.Monoid ((<>))-import Kafka.Conduit.Source-import Kafka.Conduit.Sink-import Data.Conduit-import qualified Data.Conduit.List as L+import           Control.Monad        (forM_)+import           Data.Conduit+import qualified Data.Conduit.List    as L+import           Data.Monoid          ((<>))+import           Kafka.Conduit.Sink   as KSnk+import           Kafka.Conduit.Source as KSrc  kafkaBroker :: BrokerAddress kafkaBroker = BrokerAddress "localhost:9092"@@ -17,7 +17,7 @@  -- Global consumer properties consumerProps :: ConsumerProperties-consumerProps = consumerBrokersList [kafkaBroker]+consumerProps = KSrc.brokersList [kafkaBroker]              <> groupId (ConsumerGroupId "consumer_conduit_example_group")              <> noAutoCommit @@ -28,7 +28,7 @@  -- Global producer properties producerProps :: ProducerProperties-producerProps = producerBrokersList [kafkaBroker]+producerProps = KSnk.brokersList [kafkaBroker]  main :: IO () main = do
hw-kafka-conduit.cabal view
@@ -1,5 +1,5 @@ name:                hw-kafka-conduit-version:             2.0.0+version:             2.1.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@@ -62,6 +62,7 @@   type:                 exitcode-stdio-1.0   hs-source-dirs:       test   main-is:              Spec.hs+  other-modules:        Kafka.Conduit.CombinatorSpec   ghc-options:          -Wall -threaded -rtsopts -with-rtsopts=-N   default-language:     Haskell2010   build-depends:        base
src/Kafka/Conduit/Sink.hs view
@@ -8,6 +8,7 @@ import           Control.Monad.IO.Class import           Control.Monad.Trans.Resource import           Data.Conduit+import qualified Data.Conduit.List            as L import           Kafka.Consumer  import           Kafka.Conduit.Combinators    as X@@ -92,10 +93,49 @@           res <- produceMessage prod msg           runHandler $ maybe (Right prod) Left res +-- | Ignores incoming messages and commits offsets. Commit errors are ignored.+-- This functionality should not exist as a Sink and will be removed in future versions.+-- Consider having an effect instead:+--+-- > 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 = awaitForever . const . commitAllOffsets OffsetCommit+commitOffsetsSink = flip commitOffsetsSink' (const $ pure ()) +-- | Ignores incoming messages and commits offsets. Commit errors are handled with 'handleError' effect.+-- This functionality should not exist as a Sink and will be removed in future versions.+-- Consider having an effect instead:+--+-- > 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' consumer handleError = L.mapM_ $ \_ -> do+  res <- commitAllOffsets OffsetCommit consumer+  case res of+    Nothing  -> pure ()+    Just err -> handleError err++-- | Ignores incoming messages and commits offsets, but makes sure that 'producer' has an empty outgoing queue.+-- Commit errors are ignored.+-- This functionality should not exist as a Sink and will be removed in future versions.+-- Consider having an effect instead:+--+-- > 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 c p = awaitForever . const $ do-  flushProducer p-  commitAllOffsets OffsetCommit c+flushThenCommitSink consumer producer = flushThenCommitSink' consumer producer (const $ pure ())++-- | Ignores incoming messages and commits offsets, but makes sure that 'producer' has an empty outgoing queue.+-- Commit errors are handled with 'handleError' effect.+-- This functionality should not exist as a Sink and will be removed in future versions.+-- Consider having an effect instead:+--+-- > 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' consumer producer handleError = L.mapM_ $ \_ -> do+  flushProducer producer+  res <- commitAllOffsets OffsetCommit consumer+  case res of+    Nothing  -> pure ()+    Just err -> handleError err
+ test/Kafka/Conduit/CombinatorSpec.hs view
@@ -0,0 +1,43 @@+{-# LANGUAGE ScopedTypeVariables #-}++module Kafka.Conduit.CombinatorSpec (spec) where++import           Data.Conduit+import qualified Data.Conduit.List         as CL+import           Data.List.Extra+import           Kafka.Conduit.Combinators+import           Prelude                   as P+import           Test.Hspec+import           Test.QuickCheck++{-# ANN module ("HLint: ignore Redundant do"        :: String) #-}+{-# ANN module ("HLint: ignore Reduce duplication"  :: String) #-}++spec :: Spec+spec = describe "Kafka.Conduit.UtilSpec" $ do+  describe "batchByOrFlush" $ do+    it "Should batch by i properly when Nothing is interspersed between every j elements" $ do+      forAll (choose (1 :: Int, 100)) $ \i ->+        forAll (choose (1 :: Int, 100)) $ \j ->+          testBatchBy batchByOrFlush Just [Nothing] i j+    it "Should batch by i properly when [Nothing, Nothing] is interspersed between every j elements" $ do+      forAll (choose (1 :: Int, 100)) $ \i ->+        forAll (choose (1 :: Int, 100)) $ \j ->+          testBatchBy batchByOrFlush Just [Nothing, Nothing] i j+  describe "batchByOrFlushEither" $ do+    it "Should batch by i properly when Left () is interspersed between every j elements" $ do+      forAll (choose (1 :: Int, 100)) $ \i ->+        forAll (choose (1 :: Int, 100)) $ \j ->+          testBatchBy batchByOrFlushEither Right [Left ()] i j+    it "Should batch by i properly when [Left (), Left ()] is interspersed between every j elements" $ do+      forAll (choose (1 :: Int, 100)) $ \i ->+        forAll (choose (1 :: Int, 100)) $ \j ->+          testBatchBy batchByOrFlushEither Right [Left (), Left ()] i j++testBatchBy :: (Enum a, Num a) => (BatchSize -> ConduitM b [Int] IO ()) -> (a -> b) -> [b] -> Int -> Int -> IO ()+testBatchBy batchFn constr inter i j = do+  let as = intercalate inter (chunksOf j (constr <$> [1..100]))+  xs :: [[Int]] <- runConduit $ CL.sourceList as .| batchFn (BatchSize i) .| CL.consume+  P.concat xs `shouldBe` [1..100]+  P.filter (> i) (length <$> P.init xs) `shouldBe` []+  P.filter (> j) (length <$> P.init xs) `shouldBe` []