hw-kafka-streamly 0.1.0.0 → 0.2.0.0

raw patch · 11 files changed

+727/−708 lines, 11 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

• CHANGELOG.md +23/−1
• README.md +16/−18
• hw-kafka-streamly.cabal +6/−7
• src/Kafka/Streamly/Combinators.hs +1/−1
• src/Kafka/Streamly/Fold.hs +146/−0
• src/Kafka/Streamly/Sink.hs +0/−146
• src/Kafka/Streamly/Source.hs +0/−329
• src/Kafka/Streamly/Stream.hs +329/−0
• test/Kafka/Streamly/SourceTest.hs +0/−204
• test/Kafka/Streamly/StreamTest.hs +204/−0
• test/Main.hs +2/−2

@@ -5,7 +5,29 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and this project adheres to the [Haskell Package Versioning Policy](https://pvp.haskell.org/). -## 0.1.0.0 (unreleased)+## 0.2.0.0 — 2026-05-07++### Changed (BREAKING)++- Renamed modules and functions to use streamly's vocabulary instead of+  conduit's. The functions return Streamly `Stream`s and `Fold`s, so the+  modules are now named `Kafka.Streamly.Stream` and `Kafka.Streamly.Fold`,+  and the function families are `kafkaStream*` and `kafkaFold*`.++  | Old (0.1.0.0)                  | New (0.2.0.0)                |+  |--------------------------------|------------------------------|+  | `Kafka.Streamly.Source`        | `Kafka.Streamly.Stream`      |+  | `Kafka.Streamly.Sink`          | `Kafka.Streamly.Fold`        |+  | `kafkaSource`                  | `kafkaStream`                |+  | `kafkaSourceAutoClose`         | `kafkaStreamAutoClose`       |+  | `kafkaSourceNoClose`           | `kafkaStreamNoClose`         |+  | `kafkaSink`                    | `kafkaFold`                  |+  | `kafkaBatchSink`               | `kafkaBatchFold`             |++  No backward-compatible re-exports are provided. Callers porting from+  0.1.0.0 should mechanically substitute identifiers per the table.++## 0.1.0.0 — 2026-04-17  Initial release. 

@@ -15,43 +15,41 @@  ```cabal build-depends:-  , hw-kafka-streamly  >=0.1 && <0.2+  , hw-kafka-streamly  >=0.2 && <0.3   , hw-kafka-client    >=5.3 && <6-  , streamly-core      >=0.4 && <0.5+  , streamly-core      >=0.3 && <0.5 ``` -> **Note:** `streamly-core` 0.4 is not yet published to Hackage at the time of-> writing. Until it is, depend on a `source-repository-package` for upstream-> Streamly in your `cabal.project`. The plan to land Streamly 0.4 on Hackage-> is tracked in the project's master plan-> (`docs/masterplans/1-release-hw-kafka-streamly-0.1.0.0.md`, EP-5/EP-6).+`hw-kafka-client` needs the `librdkafka` C library at build and run time+(Homebrew: `brew install librdkafka`; Debian/Ubuntu: `apt install+librdkafka-dev`).  ## Modules -- `Kafka.Streamly.Source` — consumer streams, error predicates and filters,+- `Kafka.Streamly.Stream` — consumer streams, error predicates and filters,   value-mapping helpers built on `Bifunctor`/`Bitraversable`.-- `Kafka.Streamly.Sink` — producer folds and a `withKafkaProducer` bracket+- `Kafka.Streamly.Fold` — producer folds and a `withKafkaProducer` bracket   helper. - `Kafka.Streamly.Combinators` — batching combinators and helpers that throw   `Left` values as exceptions.  ## Consuming -The source module ships three variants that differ only in how they manage+The stream module ships three variants that differ only in how they manage the underlying `KafkaConsumer`: -- `kafkaSource` — creates the consumer from `ConsumerProperties` and+- `kafkaStream` — creates the consumer from `ConsumerProperties` and   `Subscription`, closes it when the stream ends. Use this when the stream   fully owns the consumer's lifecycle.-- `kafkaSourceAutoClose` — wraps a caller-supplied `KafkaConsumer` and closes+- `kafkaStreamAutoClose` — wraps a caller-supplied `KafkaConsumer` and closes   it on stream end. Use this when the consumer is created elsewhere but its   lifetime matches the stream.-- `kafkaSourceNoClose` — wraps a caller-supplied `KafkaConsumer` and leaves+- `kafkaStreamNoClose` — wraps a caller-supplied `KafkaConsumer` and leaves   it open. Use this when the consumer outlives the stream.  ```haskell import Kafka.Consumer-import Kafka.Streamly.Source (kafkaSource, skipNonFatal)+import Kafka.Streamly.Stream (kafkaStream, skipNonFatal) import Streamly.Data.Stream qualified as Stream  main :: IO ()@@ -62,16 +60,16 @@       sub  = topics ["events"] <> offsetReset Earliest   Stream.fold (Fold.drainBy print)     . skipNonFatal-    $ kafkaSource props sub (Timeout 1000)+    $ kafkaStream props sub (Timeout 1000) ```  ## Producing -Producer sinks are Streamly `Fold`s:+Producer folds are Streamly `Fold`s:  ```haskell import Kafka.Producer-import Kafka.Streamly.Sink (kafkaSink, withKafkaProducer)+import Kafka.Streamly.Fold (kafkaFold, withKafkaProducer) import Streamly.Data.Fold qualified as Fold import Streamly.Data.Stream qualified as Stream @@ -79,7 +77,7 @@ main = do   let props = brokersList ["localhost:9092"]   result <- withKafkaProducer props $ \producer ->-    Stream.fold (kafkaSink producer)+    Stream.fold (kafkaFold producer)       . fmap mkRecord       $ Stream.fromList ["a", "b", "c"]   print result

@@ -1,12 +1,11 @@ cabal-version:   3.4 name:            hw-kafka-streamly-version:         0.1.0.0+version:         0.2.0.0 synopsis:        Streamly bindings for hw-kafka-client description:   Streamly streaming integration for hw-kafka-client, a Haskell-  binding to Apache Kafka via librdkafka. Provides composable stream-  sources for consuming and fold-based sinks for producing, with safe-  resource management.+  binding to Apache Kafka via librdkafka. Provides composable streams+  for consuming and folds for producing, with safe resource management.  license:         MIT license-file:    LICENSE@@ -37,8 +36,8 @@   import:             warnings   exposed-modules:     Kafka.Streamly.Combinators-    Kafka.Streamly.Sink-    Kafka.Streamly.Source+    Kafka.Streamly.Fold+    Kafka.Streamly.Stream    build-depends:     , base             >=4.21 && <5@@ -65,7 +64,7 @@   hs-source-dirs:     test   other-modules:     Kafka.Streamly.CombinatorsTest-    Kafka.Streamly.SourceTest+    Kafka.Streamly.StreamTest    build-depends:     , base               >=4.21 && <5

@@ -5,7 +5,7 @@ Auxiliary combinators that bridge consumer streams and producer folds: size-bounded batching with an explicit flush token, and helpers for callers who prefer exceptions over the @Either KafkaError a@ values that-'Kafka.Streamly.Source' yields.+'Kafka.Streamly.Stream' yields.  Re-exports t'Kafka.Types.BatchSize' from "Kafka.Types" for convenience. -}

@@ -0,0 +1,146 @@+{- |+Module      : Kafka.Streamly.Fold+Description : Producer folds and bracket helper for Streamly–Kafka pipelines.++Streamly expresses consumers of a stream as 'Fold' values, the dual of+@Streamly.Data.Stream.Stream@. This module provides folds that send each incoming+'Kafka.Producer.ProducerRecord' to a Kafka broker, plus a bracket helper+('withKafkaProducer') for the producer's lifecycle.++== Worked example++Fold five records through 'kafkaFold':++> import Kafka.Producer (ProducerRecord (..), TopicName (..), ProducePartition (..))+> import Kafka.Streamly.Fold (kafkaFold, withKafkaProducer)+> import Streamly.Data.Fold qualified as Fold+> import Streamly.Data.Stream qualified as Stream+>+> mkRecord :: Int -> ProducerRecord+> mkRecord i = ProducerRecord+>     { prTopic     = TopicName "example-topic"+>     , prPartition = UnassignedPartition+>     , prKey       = Nothing+>     , prValue     = Just (encodeUtf8 (pack ("payload-" <> show i)))+>     , prHeaders   = mempty+>     }+>+> main :: IO ()+> main = do+>     result <- withKafkaProducer producerProps $ \\producer ->+>         Stream.fold (kafkaFold producer) (Stream.fromList (map mkRecord [1..5]))+>     print result  -- Right Nothing on success++== Delivery semantics++A successful fold (@Nothing@) means every 'Kafka.Producer.produceMessage' call+returned without an error — which in librdkafka terms means the records were+/queued/ for delivery, not that the broker has acknowledged them. End-to-end+delivery requires a flush; 'withKafkaProducer' handles this on exit, or call+'Kafka.Producer.flushProducer' explicitly.+-}+module Kafka.Streamly.Fold (+    -- * Producer folds+    kafkaFold,+    kafkaBatchFold,++    -- * Resource management+    withKafkaProducer,+) where++import Control.Exception (bracket)+import Control.Monad.IO.Class (MonadIO, liftIO)+import Kafka.Producer (+    KafkaError,+    KafkaProducer,+    ProducerProperties,+    ProducerRecord,+    closeProducer,+    newProducer,+    produceMessage,+ )+import Streamly.Data.Fold (Fold)+import Streamly.Data.Fold qualified as Fold++{- | A 'Fold' that sends each 'ProducerRecord' to Kafka via the given producer.++Returns 'Nothing' if every 'Kafka.Producer.produceMessage' call succeeded, or+'Just' the first t'KafkaError' encountered. After an error the fold still+consumes the remaining input but sends nothing.++A 'Nothing' result means librdkafka has /queued/ all records for delivery; it+does not mean the broker has acknowledged them. For end-to-end delivery, run+the fold inside 'withKafkaProducer' (which flushes on exit) or call+'Kafka.Producer.flushProducer' directly.++@since 0.1.0.0+-}+kafkaFold ::+    (MonadIO m) =>+    KafkaProducer ->+    Fold m ProducerRecord (Maybe KafkaError)+kafkaFold producer = Fold.foldlM' step (pure Nothing)+  where+    step Nothing record = liftIO $ produceMessage producer record+    step err@(Just _) _ = pure err+{-# INLINE kafkaFold #-}++{- | A 'Fold' that sends batches of 'ProducerRecord' to Kafka.++Returns 'Nothing' if every record in every batch was accepted by+'Kafka.Producer.produceMessage', or 'Just' the first t'KafkaError'. After an+error the fold still consumes remaining input but sends nothing.++As with 'kafkaFold', 'Nothing' means records are queued in librdkafka, not+acknowledged by the broker — use 'withKafkaProducer' or+'Kafka.Producer.flushProducer' for delivery guarantees.++Note: as of @hw-kafka-client-5.3.0@ each batch is sent as individual+'Kafka.Producer.produceMessage' calls because @produceMessageBatch@ is not+exported from that release. A future version of this library may switch to a+true broker-side batch send once the upstream dependency supports it. Today+this fold is a convenience for accepting @[ProducerRecord]@ input (for example+the output of 'Kafka.Streamly.Combinators.batchByOrFlush') — it does not+reduce the number of network round-trips compared to 'kafkaFold'.++@since 0.1.0.0+-}+kafkaBatchFold ::+    (MonadIO m) =>+    KafkaProducer ->+    Fold m [ProducerRecord] (Maybe KafkaError)+kafkaBatchFold producer = Fold.foldlM' step (pure Nothing)+  where+    step Nothing batch = sendBatch batch+    step err@(Just _) _ = pure err++    sendBatch [] = pure Nothing+    sendBatch (r : rs) = do+        result <- liftIO $ produceMessage producer r+        case result of+            Nothing -> sendBatch rs+            Just err -> pure (Just err)+{-# INLINE kafkaBatchFold #-}++{- | Bracket producer creation and destruction around an action.++Creates a producer from the given 'ProducerProperties', passes it to @action@,+then closes the producer on exit. Closing a @hw-kafka-client@ producer flushes+any queued records before releasing resources, so callers do not need to flush+explicitly. Returns 'Left' if producer creation fails, otherwise 'Right' with+the action's result.++@since 0.1.0.0+-}+withKafkaProducer ::+    ProducerProperties ->+    (KafkaProducer -> IO a) ->+    IO (Either KafkaError a)+withKafkaProducer props action =+    newProducer props >>= \case+        Left err -> pure (Left err)+        Right producer ->+            bracket+                (pure producer)+                closeProducer+                (\p -> Right <$> action p)

@@ -1,146 +0,0 @@-{- |-Module      : Kafka.Streamly.Sink-Description : Producer folds and bracket helper for Streamly–Kafka pipelines.--Streamly expresses consumers of a stream as 'Fold' values, the dual of-@Streamly.Data.Stream.Stream@. This module provides folds that send each incoming-'Kafka.Producer.ProducerRecord' to a Kafka broker, plus a bracket helper-('withKafkaProducer') for the producer's lifecycle.--== Worked example--Fold five records through 'kafkaSink':--> import Kafka.Producer (ProducerRecord (..), TopicName (..), ProducePartition (..))-> import Kafka.Streamly.Sink (kafkaSink, withKafkaProducer)-> import Streamly.Data.Fold qualified as Fold-> import Streamly.Data.Stream qualified as Stream->-> mkRecord :: Int -> ProducerRecord-> mkRecord i = ProducerRecord->     { prTopic     = TopicName "example-topic"->     , prPartition = UnassignedPartition->     , prKey       = Nothing->     , prValue     = Just (encodeUtf8 (pack ("payload-" <> show i)))->     , prHeaders   = mempty->     }->-> main :: IO ()-> main = do->     result <- withKafkaProducer producerProps $ \\producer ->->         Stream.fold (kafkaSink producer) (Stream.fromList (map mkRecord [1..5]))->     print result  -- Right Nothing on success--== Delivery semantics--A successful fold (@Nothing@) means every 'Kafka.Producer.produceMessage' call-returned without an error — which in librdkafka terms means the records were-/queued/ for delivery, not that the broker has acknowledged them. End-to-end-delivery requires a flush; 'withKafkaProducer' handles this on exit, or call-'Kafka.Producer.flushProducer' explicitly.--}-module Kafka.Streamly.Sink (-    -- * Producer folds-    kafkaSink,-    kafkaBatchSink,--    -- * Resource management-    withKafkaProducer,-) where--import Control.Exception (bracket)-import Control.Monad.IO.Class (MonadIO, liftIO)-import Kafka.Producer (-    KafkaError,-    KafkaProducer,-    ProducerProperties,-    ProducerRecord,-    closeProducer,-    newProducer,-    produceMessage,- )-import Streamly.Data.Fold (Fold)-import Streamly.Data.Fold qualified as Fold--{- | A 'Fold' that sends each 'ProducerRecord' to Kafka via the given producer.--Returns 'Nothing' if every 'Kafka.Producer.produceMessage' call succeeded, or-'Just' the first t'KafkaError' encountered. After an error the fold still-consumes the remaining input but sends nothing.--A 'Nothing' result means librdkafka has /queued/ all records for delivery; it-does not mean the broker has acknowledged them. For end-to-end delivery, run-the fold inside 'withKafkaProducer' (which flushes on exit) or call-'Kafka.Producer.flushProducer' directly.--@since 0.1.0.0--}-kafkaSink ::-    (MonadIO m) =>-    KafkaProducer ->-    Fold m ProducerRecord (Maybe KafkaError)-kafkaSink producer = Fold.foldlM' step (pure Nothing)-  where-    step Nothing record = liftIO $ produceMessage producer record-    step err@(Just _) _ = pure err-{-# INLINE kafkaSink #-}--{- | A 'Fold' that sends batches of 'ProducerRecord' to Kafka.--Returns 'Nothing' if every record in every batch was accepted by-'Kafka.Producer.produceMessage', or 'Just' the first t'KafkaError'. After an-error the fold still consumes remaining input but sends nothing.--As with 'kafkaSink', 'Nothing' means records are queued in librdkafka, not-acknowledged by the broker — use 'withKafkaProducer' or-'Kafka.Producer.flushProducer' for delivery guarantees.--Note: as of @hw-kafka-client-5.3.0@ each batch is sent as individual-'Kafka.Producer.produceMessage' calls because @produceMessageBatch@ is not-exported from that release. A future version of this library may switch to a-true broker-side batch send once the upstream dependency supports it. Today-this fold is a convenience for accepting @[ProducerRecord]@ input (for example-the output of 'Kafka.Streamly.Combinators.batchByOrFlush') — it does not-reduce the number of network round-trips compared to 'kafkaSink'.--@since 0.1.0.0--}-kafkaBatchSink ::-    (MonadIO m) =>-    KafkaProducer ->-    Fold m [ProducerRecord] (Maybe KafkaError)-kafkaBatchSink producer = Fold.foldlM' step (pure Nothing)-  where-    step Nothing batch = sendBatch batch-    step err@(Just _) _ = pure err--    sendBatch [] = pure Nothing-    sendBatch (r : rs) = do-        result <- liftIO $ produceMessage producer r-        case result of-            Nothing -> sendBatch rs-            Just err -> pure (Just err)-{-# INLINE kafkaBatchSink #-}--{- | Bracket producer creation and destruction around an action.--Creates a producer from the given 'ProducerProperties', passes it to @action@,-then closes the producer on exit. Closing a @hw-kafka-client@ producer flushes-any queued records before releasing resources, so callers do not need to flush-explicitly. Returns 'Left' if producer creation fails, otherwise 'Right' with-the action's result.--@since 0.1.0.0--}-withKafkaProducer ::-    ProducerProperties ->-    (KafkaProducer -> IO a) ->-    IO (Either KafkaError a)-withKafkaProducer props action =-    newProducer props >>= \case-        Left err -> pure (Left err)-        Right producer ->-            bracket-                (pure producer)-                closeProducer-                (\p -> Right <$> action p)

@@ -1,329 +0,0 @@-{- |-Module      : Kafka.Streamly.Source-Description : Consumer stream sources and helpers for Streamly–Kafka pipelines.--Stream sources backed by a @hw-kafka-client@ 'KafkaConsumer'. Each source-yields @Either t'Kafka.Consumer.KafkaError' ('ConsumerRecord' (Maybe ByteString) (Maybe ByteString))@ —-errors are kept in-band rather than raised as exceptions, matching the shape-exposed by 'Kafka.Consumer.pollMessage'.--== Three tiers of resource management--* 'kafkaSource' — fully managed: creates a consumer, polls, closes on stream-  termination. If consumer creation fails the t'KafkaError' is thrown as an-  exception (see the function's docstring for how this differs from-  @hw-kafka-conduit@).-* 'kafkaSourceAutoClose' — takes a pre-built consumer, polls, closes on-  termination. The caller owns creation, the stream owns destruction.-* 'kafkaSourceNoClose' — polls a consumer the caller fully owns. The stream-  never touches the consumer's lifecycle.--== Worked example--Consume five records from a topic, skipping non-fatal errors:--> import Kafka.Consumer->     ( ConsumerGroupId (..), ConsumerRecord, KafkaError->     , OffsetReset (..), brokersList, groupId, offsetReset, topics->     )-> import Kafka.Streamly.Source (kafkaSource, skipNonFatal)-> import Streamly.Data.Fold qualified as Fold-> import Streamly.Data.Stream qualified as Stream->-> main :: IO ()-> main = do->     let props = brokersList ["localhost:9092"]->             <> groupId (ConsumerGroupId "example-group")->         sub  = topics ["example-topic"] <> offsetReset Earliest->     Stream.fold (Fold.drainMapM print) $->         Stream.take 5 $->             skipNonFatal $->                 kafkaSource props sub (Timeout 1000)--== Transforming the stream--Because the element type is @Either KafkaError (ConsumerRecord k v)@, the-idiomatic way to map the inner record value is a double @fmap@:--> fmap (fmap (\\r -> r { crValue = Nothing })) stream--The helpers 'mapFirst', 'mapValue', and 'bimapValue' save one level of wrapping-by lifting a function directly into the outer stream element — see their-docstrings.--}-module Kafka.Streamly.Source (-    -- * Stream sources-    kafkaSource,-    kafkaSourceAutoClose,-    kafkaSourceNoClose,--    -- * Error predicates-    isFatal,-    isPollTimeout,-    isPartitionEOF,--    -- * Error filters-    skipNonFatal,-    skipNonFatalExcept,--    -- * Value mapping-    mapFirst,-    mapValue,-    bimapValue,-) where--import Control.Exception (throwIO)-import Control.Monad.Catch (MonadCatch)-import Control.Monad.IO.Class (MonadIO, liftIO)-import Data.Bifunctor (Bifunctor, bimap, first)-import Data.ByteString qualified as BS-import Kafka.Consumer (-    ConsumerProperties,-    ConsumerRecord,-    KafkaConsumer,-    KafkaError (..),-    RdKafkaRespErrT (..),-    Subscription,-    Timeout,-    closeConsumer,-    newConsumer,-    pollMessage,- )-import Streamly.Data.Stream (Stream)-import Streamly.Data.Stream qualified as Stream--{- | Create a 'Stream' for a given 'KafkaConsumer'.--The consumer will /not/ be closed when the stream ends; the caller owns its-lifecycle. The stream terminates the first time 'pollMessage' returns a fatal-t'KafkaError' (see 'isFatal'); non-fatal errors are yielded in-band as @Left@-values and polling continues.--@since 0.1.0.0--}-kafkaSourceNoClose ::-    (MonadIO m) =>-    KafkaConsumer ->-    Timeout ->-    Stream m (Either KafkaError (ConsumerRecord (Maybe BS.ByteString) (Maybe BS.ByteString)))-kafkaSourceNoClose consumer timeout =-    Stream.unfoldrM step True-  where-    step False = pure Nothing-    step True = do-        msg <- liftIO $ pollMessage consumer timeout-        case msg of-            Left err | isFatal err -> pure $ Just (Left err, False)-            _ -> pure $ Just (msg, True)-{-# INLINE kafkaSourceNoClose #-}--{- | Create a 'Stream' for a given 'KafkaConsumer'.--The consumer is passed in by the caller but will be closed automatically when-the stream ends (via 'Streamly.Data.Stream.bracketIO'). Use this when you want-the stream to own destruction but not creation.--@since 0.1.0.0--}-kafkaSourceAutoClose ::-    (MonadIO m, MonadCatch m) =>-    KafkaConsumer ->-    Timeout ->-    Stream m (Either KafkaError (ConsumerRecord (Maybe BS.ByteString) (Maybe BS.ByteString)))-kafkaSourceAutoClose consumer timeout =-    Stream.bracketIO-        (pure consumer)-        (\c -> () <$ closeConsumer c)-        (\c -> kafkaSourceNoClose c timeout)-{-# INLINE kafkaSourceAutoClose #-}--{- | Create a fully managed consumer 'Stream' from 'ConsumerProperties' and a-'Subscription'.--The stream creates the consumer on first demand, polls messages, and closes-the consumer when the stream ends.--If consumer creation fails, the t'KafkaError' is thrown as an exception (via-'throwIO'). This diverges from @hw-kafka-conduit@'s @kafkaSource@, which-yields @Left err@ as the first stream value and then terminates. Callers-migrating from conduit who prefer the in-band error should create the-consumer manually with 'Kafka.Consumer.newConsumer' and, on @Right c@, pass-@c@ to 'kafkaSourceAutoClose'.--Per-poll errors (non-fatal or fatal) are yielded in-band as @Left@ values,-as with the other sources in this module.--@since 0.1.0.0--}-kafkaSource ::-    (MonadIO m, MonadCatch m) =>-    ConsumerProperties ->-    Subscription ->-    Timeout ->-    Stream m (Either KafkaError (ConsumerRecord (Maybe BS.ByteString) (Maybe BS.ByteString)))-kafkaSource props sub timeout =-    Stream.bracketIO-        ( newConsumer props sub >>= \case-            Left err -> throwIO err-            Right c -> pure c-        )-        (\c -> () <$ closeConsumer c)-        (\c -> kafkaSourceNoClose c timeout)-{-# INLINE kafkaSource #-}------------------------------------------------------------------------------------ Error predicates----------------------------------------------------------------------------------{- | Checks if the error is fatal in a way that it doesn't make sense to retry-after or is unsafe to ignore.--@since 0.1.0.0--}-isFatal :: KafkaError -> Bool-isFatal = \case-    KafkaUnknownConfigurationKey _ -> True-    KafkaInvalidConfigurationValue _ -> True-    KafkaBadConfiguration -> True-    KafkaBadSpecification _ -> True-    KafkaResponseError RdKafkaRespErrDestroy -> True-    KafkaResponseError RdKafkaRespErrFail -> True-    KafkaResponseError RdKafkaRespErrInvalidArg -> True-    KafkaResponseError RdKafkaRespErrSsl -> True-    KafkaResponseError RdKafkaRespErrUnknownProtocol -> True-    KafkaResponseError RdKafkaRespErrNotImplemented -> True-    KafkaResponseError RdKafkaRespErrAuthentication -> True-    KafkaResponseError RdKafkaRespErrInconsistentGroupProtocol -> True-    KafkaResponseError RdKafkaRespErrTopicAuthorizationFailed -> True-    KafkaResponseError RdKafkaRespErrGroupAuthorizationFailed -> True-    KafkaResponseError RdKafkaRespErrClusterAuthorizationFailed -> True-    KafkaResponseError RdKafkaRespErrUnsupportedSaslMechanism -> True-    KafkaResponseError RdKafkaRespErrIllegalSaslState -> True-    KafkaResponseError RdKafkaRespErrUnsupportedVersion -> True-    _ -> False-{-# INLINE isFatal #-}--{- | Checks if the error is a poll timeout ('RdKafkaRespErrTimedOut').-Timeout errors are not fatal and occur when 'pollMessage' has no messages-to return within the specified timeout.--@since 0.1.0.0--}-isPollTimeout :: KafkaError -> Bool-isPollTimeout e = KafkaResponseError RdKafkaRespErrTimedOut == e-{-# INLINE isPollTimeout #-}--{- | Checks if the error indicates reaching the end of a partition-('RdKafkaRespErrPartitionEof'). Not fatal; occurs every time a consumer-reaches the current end of a partition.--@since 0.1.0.0--}-isPartitionEOF :: KafkaError -> Bool-isPartitionEOF e = KafkaResponseError RdKafkaRespErrPartitionEof == e-{-# INLINE isPartitionEOF #-}------------------------------------------------------------------------------------ Error filters----------------------------------------------------------------------------------{- | Filter out non-fatal errors, keeping only fatal errors and successful-records.--This drops every t'KafkaError' for which 'isFatal' returns 'False', which-includes poll timeouts ('RdKafkaRespErrTimedOut') /and/ partition-EOF markers-('RdKafkaRespErrPartitionEof'). If your consumer relies on observing EOFs —-for example, to terminate a bounded read — use 'skipNonFatalExcept' with-'isPartitionEOF' instead.--@since 0.1.0.0--}-skipNonFatal ::-    (Monad m) =>-    Stream m (Either KafkaError b) ->-    Stream m (Either KafkaError b)-skipNonFatal = Stream.filter (either isFatal (const True))-{-# INLINE skipNonFatal #-}--{- | Filter out non-fatal errors except those matching any of the given-predicates. Fatal errors always pass through.--> skipNonFatalExcept [isPollTimeout, isPartitionEOF]--@since 0.1.0.0--}-skipNonFatalExcept ::-    (Monad m) =>-    [KafkaError -> Bool] ->-    Stream m (Either KafkaError b) ->-    Stream m (Either KafkaError b)-skipNonFatalExcept fs =-    let fun e = or $ (\f -> f e) <$> (isFatal : fs)-     in Stream.filter (either fun (const True))-{-# INLINE skipNonFatalExcept #-}------------------------------------------------------------------------------------ Value mapping----------------------------------------------------------------------------------{- | Lift a function into the left position of a 'Bifunctor' stream element.--Given a stream of @t k v@ (for example @Either KafkaError (ConsumerRecord k v)@),-transform the left component. To map the key of a 'ConsumerRecord' inside an-@Either@ the caller still has to compose with an outer 'fmap':--> mapFirst (fmap prefixKey) :: Stream m (Either e (ConsumerRecord k v)) -> ...--@since 0.1.0.0--}-mapFirst ::-    (Bifunctor t, Monad m) =>-    (k -> k') ->-    Stream m (t k v) ->-    Stream m (t k' v)-mapFirst f = fmap (first f)-{-# INLINE mapFirst #-}--{- | Lift a function into the 'Functor' position of a stream element.--For @Stream m (Either e (ConsumerRecord k v))@, @mapValue f@ rewrites the-@Right@ side, leaving errors untouched. To transform the record /value/-itself the caller composes with another 'fmap':--> mapValue (fmap uppercase) :: Stream m (Either e (ConsumerRecord k v)) -> ...--@since 0.1.0.0--}-mapValue ::-    (Functor t, Monad m) =>-    (v -> v') ->-    Stream m (t v) ->-    Stream m (t v')-mapValue f = fmap (fmap f)-{-# INLINE mapValue #-}--{- | Lift a pair of functions into the two positions of a 'Bifunctor' stream-element.--Acts on the outer element. For @Stream m (Either KafkaError x)@ this maps-both the error and the success:--> bimapValue renderError handleSuccess->     :: Stream m (Either KafkaError x) -> Stream m (Either String y)--To bimap the key and value /inside/ an outer 'Either', wrap with 'fmap':--> mapValue (bimap prefixKey uppercase)->     :: Stream m (Either e (ConsumerRecord k v))->     -> Stream m (Either e (ConsumerRecord k' v'))--@since 0.1.0.0--}-bimapValue ::-    (Bifunctor t, Monad m) =>-    (k -> k') ->-    (v -> v') ->-    Stream m (t k v) ->-    Stream m (t k' v')-bimapValue f g = fmap (bimap f g)-{-# INLINE bimapValue #-}

@@ -0,0 +1,329 @@+{- |+Module      : Kafka.Streamly.Stream+Description : Consumer streams and helpers for Streamly–Kafka pipelines.++Streams backed by a @hw-kafka-client@ 'KafkaConsumer'. Each stream+yields @Either t'Kafka.Consumer.KafkaError' ('ConsumerRecord' (Maybe ByteString) (Maybe ByteString))@ —+errors are kept in-band rather than raised as exceptions, matching the shape+exposed by 'Kafka.Consumer.pollMessage'.++== Three tiers of resource management++* 'kafkaStream' — fully managed: creates a consumer, polls, closes on stream+  termination. If consumer creation fails the t'KafkaError' is thrown as an+  exception (see the function's docstring for how this differs from+  @hw-kafka-conduit@).+* 'kafkaStreamAutoClose' — takes a pre-built consumer, polls, closes on+  termination. The caller owns creation, the stream owns destruction.+* 'kafkaStreamNoClose' — polls a consumer the caller fully owns. The stream+  never touches the consumer's lifecycle.++== Worked example++Consume five records from a topic, skipping non-fatal errors:++> import Kafka.Consumer+>     ( ConsumerGroupId (..), ConsumerRecord, KafkaError+>     , OffsetReset (..), brokersList, groupId, offsetReset, topics+>     )+> import Kafka.Streamly.Stream (kafkaStream, skipNonFatal)+> import Streamly.Data.Fold qualified as Fold+> import Streamly.Data.Stream qualified as Stream+>+> main :: IO ()+> main = do+>     let props = brokersList ["localhost:9092"]+>             <> groupId (ConsumerGroupId "example-group")+>         sub  = topics ["example-topic"] <> offsetReset Earliest+>     Stream.fold (Fold.drainMapM print) $+>         Stream.take 5 $+>             skipNonFatal $+>                 kafkaStream props sub (Timeout 1000)++== Transforming the stream++Because the element type is @Either KafkaError (ConsumerRecord k v)@, the+idiomatic way to map the inner record value is a double @fmap@:++> fmap (fmap (\\r -> r { crValue = Nothing })) stream++The helpers 'mapFirst', 'mapValue', and 'bimapValue' save one level of wrapping+by lifting a function directly into the outer stream element — see their+docstrings.+-}+module Kafka.Streamly.Stream (+    -- * Streams+    kafkaStream,+    kafkaStreamAutoClose,+    kafkaStreamNoClose,++    -- * Error predicates+    isFatal,+    isPollTimeout,+    isPartitionEOF,++    -- * Error filters+    skipNonFatal,+    skipNonFatalExcept,++    -- * Value mapping+    mapFirst,+    mapValue,+    bimapValue,+) where++import Control.Exception (throwIO)+import Control.Monad.Catch (MonadCatch)+import Control.Monad.IO.Class (MonadIO, liftIO)+import Data.Bifunctor (Bifunctor, bimap, first)+import Data.ByteString qualified as BS+import Kafka.Consumer (+    ConsumerProperties,+    ConsumerRecord,+    KafkaConsumer,+    KafkaError (..),+    RdKafkaRespErrT (..),+    Subscription,+    Timeout,+    closeConsumer,+    newConsumer,+    pollMessage,+ )+import Streamly.Data.Stream (Stream)+import Streamly.Data.Stream qualified as Stream++{- | Create a 'Stream' for a given 'KafkaConsumer'.++The consumer will /not/ be closed when the stream ends; the caller owns its+lifecycle. The stream terminates the first time 'pollMessage' returns a fatal+t'KafkaError' (see 'isFatal'); non-fatal errors are yielded in-band as @Left@+values and polling continues.++@since 0.1.0.0+-}+kafkaStreamNoClose ::+    (MonadIO m) =>+    KafkaConsumer ->+    Timeout ->+    Stream m (Either KafkaError (ConsumerRecord (Maybe BS.ByteString) (Maybe BS.ByteString)))+kafkaStreamNoClose consumer timeout =+    Stream.unfoldrM step True+  where+    step False = pure Nothing+    step True = do+        msg <- liftIO $ pollMessage consumer timeout+        case msg of+            Left err | isFatal err -> pure $ Just (Left err, False)+            _ -> pure $ Just (msg, True)+{-# INLINE kafkaStreamNoClose #-}++{- | Create a 'Stream' for a given 'KafkaConsumer'.++The consumer is passed in by the caller but will be closed automatically when+the stream ends (via 'Streamly.Data.Stream.bracketIO'). Use this when you want+the stream to own destruction but not creation.++@since 0.1.0.0+-}+kafkaStreamAutoClose ::+    (MonadIO m, MonadCatch m) =>+    KafkaConsumer ->+    Timeout ->+    Stream m (Either KafkaError (ConsumerRecord (Maybe BS.ByteString) (Maybe BS.ByteString)))+kafkaStreamAutoClose consumer timeout =+    Stream.bracketIO+        (pure consumer)+        (\c -> () <$ closeConsumer c)+        (\c -> kafkaStreamNoClose c timeout)+{-# INLINE kafkaStreamAutoClose #-}++{- | Create a fully managed consumer 'Stream' from 'ConsumerProperties' and a+'Subscription'.++The stream creates the consumer on first demand, polls messages, and closes+the consumer when the stream ends.++If consumer creation fails, the t'KafkaError' is thrown as an exception (via+'throwIO'). This diverges from @hw-kafka-conduit@'s @kafkaSource@, which+yields @Left err@ as the first stream value and then terminates. Callers+migrating from conduit who prefer the in-band error should create the+consumer manually with 'Kafka.Consumer.newConsumer' and, on @Right c@, pass+@c@ to 'kafkaStreamAutoClose'.++Per-poll errors (non-fatal or fatal) are yielded in-band as @Left@ values,+as with the other streams in this module.++@since 0.1.0.0+-}+kafkaStream ::+    (MonadIO m, MonadCatch m) =>+    ConsumerProperties ->+    Subscription ->+    Timeout ->+    Stream m (Either KafkaError (ConsumerRecord (Maybe BS.ByteString) (Maybe BS.ByteString)))+kafkaStream props sub timeout =+    Stream.bracketIO+        ( newConsumer props sub >>= \case+            Left err -> throwIO err+            Right c -> pure c+        )+        (\c -> () <$ closeConsumer c)+        (\c -> kafkaStreamNoClose c timeout)+{-# INLINE kafkaStream #-}++-------------------------------------------------------------------------------+-- Error predicates+-------------------------------------------------------------------------------++{- | Checks if the error is fatal in a way that it doesn't make sense to retry+after or is unsafe to ignore.++@since 0.1.0.0+-}+isFatal :: KafkaError -> Bool+isFatal = \case+    KafkaUnknownConfigurationKey _ -> True+    KafkaInvalidConfigurationValue _ -> True+    KafkaBadConfiguration -> True+    KafkaBadSpecification _ -> True+    KafkaResponseError RdKafkaRespErrDestroy -> True+    KafkaResponseError RdKafkaRespErrFail -> True+    KafkaResponseError RdKafkaRespErrInvalidArg -> True+    KafkaResponseError RdKafkaRespErrSsl -> True+    KafkaResponseError RdKafkaRespErrUnknownProtocol -> True+    KafkaResponseError RdKafkaRespErrNotImplemented -> True+    KafkaResponseError RdKafkaRespErrAuthentication -> True+    KafkaResponseError RdKafkaRespErrInconsistentGroupProtocol -> True+    KafkaResponseError RdKafkaRespErrTopicAuthorizationFailed -> True+    KafkaResponseError RdKafkaRespErrGroupAuthorizationFailed -> True+    KafkaResponseError RdKafkaRespErrClusterAuthorizationFailed -> True+    KafkaResponseError RdKafkaRespErrUnsupportedSaslMechanism -> True+    KafkaResponseError RdKafkaRespErrIllegalSaslState -> True+    KafkaResponseError RdKafkaRespErrUnsupportedVersion -> True+    _ -> False+{-# INLINE isFatal #-}++{- | Checks if the error is a poll timeout ('RdKafkaRespErrTimedOut').+Timeout errors are not fatal and occur when 'pollMessage' has no messages+to return within the specified timeout.++@since 0.1.0.0+-}+isPollTimeout :: KafkaError -> Bool+isPollTimeout e = KafkaResponseError RdKafkaRespErrTimedOut == e+{-# INLINE isPollTimeout #-}++{- | Checks if the error indicates reaching the end of a partition+('RdKafkaRespErrPartitionEof'). Not fatal; occurs every time a consumer+reaches the current end of a partition.++@since 0.1.0.0+-}+isPartitionEOF :: KafkaError -> Bool+isPartitionEOF e = KafkaResponseError RdKafkaRespErrPartitionEof == e+{-# INLINE isPartitionEOF #-}++-------------------------------------------------------------------------------+-- Error filters+-------------------------------------------------------------------------------++{- | Filter out non-fatal errors, keeping only fatal errors and successful+records.++This drops every t'KafkaError' for which 'isFatal' returns 'False', which+includes poll timeouts ('RdKafkaRespErrTimedOut') /and/ partition-EOF markers+('RdKafkaRespErrPartitionEof'). If your consumer relies on observing EOFs —+for example, to terminate a bounded read — use 'skipNonFatalExcept' with+'isPartitionEOF' instead.++@since 0.1.0.0+-}+skipNonFatal ::+    (Monad m) =>+    Stream m (Either KafkaError b) ->+    Stream m (Either KafkaError b)+skipNonFatal = Stream.filter (either isFatal (const True))+{-# INLINE skipNonFatal #-}++{- | Filter out non-fatal errors except those matching any of the given+predicates. Fatal errors always pass through.++> skipNonFatalExcept [isPollTimeout, isPartitionEOF]++@since 0.1.0.0+-}+skipNonFatalExcept ::+    (Monad m) =>+    [KafkaError -> Bool] ->+    Stream m (Either KafkaError b) ->+    Stream m (Either KafkaError b)+skipNonFatalExcept fs =+    let fun e = or $ (\f -> f e) <$> (isFatal : fs)+     in Stream.filter (either fun (const True))+{-# INLINE skipNonFatalExcept #-}++-------------------------------------------------------------------------------+-- Value mapping+-------------------------------------------------------------------------------++{- | Lift a function into the left position of a 'Bifunctor' stream element.++Given a stream of @t k v@ (for example @Either KafkaError (ConsumerRecord k v)@),+transform the left component. To map the key of a 'ConsumerRecord' inside an+@Either@ the caller still has to compose with an outer 'fmap':++> mapFirst (fmap prefixKey) :: Stream m (Either e (ConsumerRecord k v)) -> ...++@since 0.1.0.0+-}+mapFirst ::+    (Bifunctor t, Monad m) =>+    (k -> k') ->+    Stream m (t k v) ->+    Stream m (t k' v)+mapFirst f = fmap (first f)+{-# INLINE mapFirst #-}++{- | Lift a function into the 'Functor' position of a stream element.++For @Stream m (Either e (ConsumerRecord k v))@, @mapValue f@ rewrites the+@Right@ side, leaving errors untouched. To transform the record /value/+itself the caller composes with another 'fmap':++> mapValue (fmap uppercase) :: Stream m (Either e (ConsumerRecord k v)) -> ...++@since 0.1.0.0+-}+mapValue ::+    (Functor t, Monad m) =>+    (v -> v') ->+    Stream m (t v) ->+    Stream m (t v')+mapValue f = fmap (fmap f)+{-# INLINE mapValue #-}++{- | Lift a pair of functions into the two positions of a 'Bifunctor' stream+element.++Acts on the outer element. For @Stream m (Either KafkaError x)@ this maps+both the error and the success:++> bimapValue renderError handleSuccess+>     :: Stream m (Either KafkaError x) -> Stream m (Either String y)++To bimap the key and value /inside/ an outer 'Either', wrap with 'fmap':++> mapValue (bimap prefixKey uppercase)+>     :: Stream m (Either e (ConsumerRecord k v))+>     -> Stream m (Either e (ConsumerRecord k' v'))++@since 0.1.0.0+-}+bimapValue ::+    (Bifunctor t, Monad m) =>+    (k -> k') ->+    (v -> v') ->+    Stream m (t k v) ->+    Stream m (t k' v')+bimapValue f g = fmap (bimap f g)+{-# INLINE bimapValue #-}

@@ -1,204 +0,0 @@-module Kafka.Streamly.SourceTest (tests) where--import Kafka.Consumer (KafkaError (..), RdKafkaRespErrT (..))-import Kafka.Streamly.Source (-    isFatal,-    isPartitionEOF,-    isPollTimeout,-    skipNonFatal,-    skipNonFatalExcept,- )-import Streamly.Data.Fold qualified as Fold-import Streamly.Data.Stream (Stream)-import Streamly.Data.Stream qualified as Stream-import Test.Tasty (TestTree, testGroup)-import Test.Tasty.HUnit (Assertion, testCase, (@?=))--tests :: TestTree-tests =-    testGroup-        "Source"-        [ testGroup "isFatal" isFatalTests-        , testGroup "isPollTimeout" isPollTimeoutTests-        , testGroup "isPartitionEOF" isPartitionEOFTests-        , testGroup "skipNonFatal" skipNonFatalTests-        , testGroup "skipNonFatalExcept" skipNonFatalExceptTests-        ]------------------------------------------------------------------------------------ isFatal----------------------------------------------------------------------------------fatalErrors :: [(String, KafkaError)]-fatalErrors =-    [ ("KafkaUnknownConfigurationKey", KafkaUnknownConfigurationKey "")-    , ("KafkaInvalidConfigurationValue", KafkaInvalidConfigurationValue "")-    , ("KafkaBadConfiguration", KafkaBadConfiguration)-    , ("KafkaBadSpecification", KafkaBadSpecification "")-    , ("RdKafkaRespErrDestroy", KafkaResponseError RdKafkaRespErrDestroy)-    , ("RdKafkaRespErrFail", KafkaResponseError RdKafkaRespErrFail)-    , ("RdKafkaRespErrInvalidArg", KafkaResponseError RdKafkaRespErrInvalidArg)-    , ("RdKafkaRespErrSsl", KafkaResponseError RdKafkaRespErrSsl)-    , ("RdKafkaRespErrUnknownProtocol", KafkaResponseError RdKafkaRespErrUnknownProtocol)-    , ("RdKafkaRespErrNotImplemented", KafkaResponseError RdKafkaRespErrNotImplemented)-    , ("RdKafkaRespErrAuthentication", KafkaResponseError RdKafkaRespErrAuthentication)-    , ("RdKafkaRespErrInconsistentGroupProtocol", KafkaResponseError RdKafkaRespErrInconsistentGroupProtocol)-    , ("RdKafkaRespErrTopicAuthorizationFailed", KafkaResponseError RdKafkaRespErrTopicAuthorizationFailed)-    , ("RdKafkaRespErrGroupAuthorizationFailed", KafkaResponseError RdKafkaRespErrGroupAuthorizationFailed)-    , ("RdKafkaRespErrClusterAuthorizationFailed", KafkaResponseError RdKafkaRespErrClusterAuthorizationFailed)-    , ("RdKafkaRespErrUnsupportedSaslMechanism", KafkaResponseError RdKafkaRespErrUnsupportedSaslMechanism)-    , ("RdKafkaRespErrIllegalSaslState", KafkaResponseError RdKafkaRespErrIllegalSaslState)-    , ("RdKafkaRespErrUnsupportedVersion", KafkaResponseError RdKafkaRespErrUnsupportedVersion)-    ]--nonFatalErrors :: [(String, KafkaError)]-nonFatalErrors =-    [ ("RdKafkaRespErrTimedOut", KafkaResponseError RdKafkaRespErrTimedOut)-    , ("RdKafkaRespErrPartitionEof", KafkaResponseError RdKafkaRespErrPartitionEof)-    , ("RdKafkaRespErrNoError", KafkaResponseError RdKafkaRespErrNoError)-    ]--isFatalTests :: [TestTree]-isFatalTests =-    [ testCase ("fatal: " <> label) (isFatal err @?= True)-    | (label, err) <- fatalErrors-    ]-        <> [ testCase ("non-fatal: " <> label) (isFatal err @?= False)-           | (label, err) <- nonFatalErrors-           ]------------------------------------------------------------------------------------ isPollTimeout----------------------------------------------------------------------------------isPollTimeoutTests :: [TestTree]-isPollTimeoutTests =-    [ testCase "matches RdKafkaRespErrTimedOut" $-        isPollTimeout (KafkaResponseError RdKafkaRespErrTimedOut) @?= True-    , testCase "rejects RdKafkaRespErrPartitionEof" $-        isPollTimeout (KafkaResponseError RdKafkaRespErrPartitionEof) @?= False-    , testCase "rejects RdKafkaRespErrAuthentication" $-        isPollTimeout (KafkaResponseError RdKafkaRespErrAuthentication) @?= False-    , testCase "rejects KafkaBadConfiguration" $-        isPollTimeout KafkaBadConfiguration @?= False-    ]------------------------------------------------------------------------------------ isPartitionEOF----------------------------------------------------------------------------------isPartitionEOFTests :: [TestTree]-isPartitionEOFTests =-    [ testCase "matches RdKafkaRespErrPartitionEof" $-        isPartitionEOF (KafkaResponseError RdKafkaRespErrPartitionEof) @?= True-    , testCase "rejects RdKafkaRespErrTimedOut" $-        isPartitionEOF (KafkaResponseError RdKafkaRespErrTimedOut) @?= False-    , testCase "rejects RdKafkaRespErrAuthentication" $-        isPartitionEOF (KafkaResponseError RdKafkaRespErrAuthentication) @?= False-    , testCase "rejects KafkaBadConfiguration" $-        isPartitionEOF KafkaBadConfiguration @?= False-    ]------------------------------------------------------------------------------------ skipNonFatal / skipNonFatalExcept----------------------------------------------------------------------------------runStream ::-    (Stream IO a -> Stream IO b) ->-    [a] ->-    IO [b]-runStream f xs = Stream.fold Fold.toList (f (Stream.fromList xs))--timedOut, partitionEof, authentication :: KafkaError-timedOut = KafkaResponseError RdKafkaRespErrTimedOut-partitionEof = KafkaResponseError RdKafkaRespErrPartitionEof-authentication = KafkaResponseError RdKafkaRespErrAuthentication--skipNonFatalTests :: [TestTree]-skipNonFatalTests =-    [ testCase "drops poll-timeout, keeps authentication, keeps all Rights" dropsAndKeeps-    , testCase "passes all Rights through unchanged" passesAllRights-    , testCase "drops non-fatal Lefts even with empty extension list" dropsNonFatal-    , testCase "drops partition-EOF (non-fatal)" dropsPartitionEof-    ]-  where-    dropsAndKeeps :: Assertion-    dropsAndKeeps = do-        let input =-                [ Right (1 :: Int)-                , Left timedOut-                , Right 2-                , Left authentication-                , Right 3-                ]-        result <- runStream skipNonFatal input-        result @?= [Right 1, Right 2, Left authentication, Right 3]--    passesAllRights :: Assertion-    passesAllRights = do-        let input = [Right (n :: Int) | n <- [1 .. 5]]-        result <- runStream skipNonFatal input-        result @?= input--    dropsNonFatal :: Assertion-    dropsNonFatal = do-        let input :: [Either KafkaError Int]-            input =-                [ Left timedOut-                , Left partitionEof-                , Left (KafkaResponseError RdKafkaRespErrNoError)-                ]-        result <- runStream skipNonFatal input-        result @?= []--    dropsPartitionEof :: Assertion-    dropsPartitionEof = do-        let input =-                [ Right (1 :: Int)-                , Left partitionEof-                , Right 2-                ]-        result <- runStream skipNonFatal input-        result @?= [Right 1, Right 2]--skipNonFatalExceptTests :: [TestTree]-skipNonFatalExceptTests =-    [ testCase "keeps poll-timeout when isPollTimeout is in the list" keepsTimeouts-    , testCase "keeps partition-EOF when isPartitionEOF is in the list" keepsPartitionEof-    , testCase "empty predicate list behaves like skipNonFatal" emptyListEquivalent-    ]-  where-    keepsTimeouts :: Assertion-    keepsTimeouts = do-        let input =-                [ Right (1 :: Int)-                , Left timedOut-                , Right 2-                , Left authentication-                , Right 3-                ]-        result <- runStream (skipNonFatalExcept [isPollTimeout]) input-        result @?= [Right 1, Left timedOut, Right 2, Left authentication, Right 3]--    keepsPartitionEof :: Assertion-    keepsPartitionEof = do-        let input =-                [ Right (1 :: Int)-                , Left partitionEof-                , Left timedOut-                , Right 2-                ]-        result <- runStream (skipNonFatalExcept [isPartitionEOF]) input-        result @?= [Right 1, Left partitionEof, Right 2]--    emptyListEquivalent :: Assertion-    emptyListEquivalent = do-        let input =-                [ Right (1 :: Int)-                , Left timedOut-                , Left authentication-                , Right 2-                ]-        viaExcept <- runStream (skipNonFatalExcept []) input-        viaPlain <- runStream skipNonFatal input-        viaExcept @?= viaPlain

@@ -0,0 +1,204 @@+module Kafka.Streamly.StreamTest (tests) where++import Kafka.Consumer (KafkaError (..), RdKafkaRespErrT (..))+import Kafka.Streamly.Stream (+    isFatal,+    isPartitionEOF,+    isPollTimeout,+    skipNonFatal,+    skipNonFatalExcept,+ )+import Streamly.Data.Fold qualified as Fold+import Streamly.Data.Stream (Stream)+import Streamly.Data.Stream qualified as Stream+import Test.Tasty (TestTree, testGroup)+import Test.Tasty.HUnit (Assertion, testCase, (@?=))++tests :: TestTree+tests =+    testGroup+        "Stream"+        [ testGroup "isFatal" isFatalTests+        , testGroup "isPollTimeout" isPollTimeoutTests+        , testGroup "isPartitionEOF" isPartitionEOFTests+        , testGroup "skipNonFatal" skipNonFatalTests+        , testGroup "skipNonFatalExcept" skipNonFatalExceptTests+        ]++-------------------------------------------------------------------------------+-- isFatal+-------------------------------------------------------------------------------++fatalErrors :: [(String, KafkaError)]+fatalErrors =+    [ ("KafkaUnknownConfigurationKey", KafkaUnknownConfigurationKey "")+    , ("KafkaInvalidConfigurationValue", KafkaInvalidConfigurationValue "")+    , ("KafkaBadConfiguration", KafkaBadConfiguration)+    , ("KafkaBadSpecification", KafkaBadSpecification "")+    , ("RdKafkaRespErrDestroy", KafkaResponseError RdKafkaRespErrDestroy)+    , ("RdKafkaRespErrFail", KafkaResponseError RdKafkaRespErrFail)+    , ("RdKafkaRespErrInvalidArg", KafkaResponseError RdKafkaRespErrInvalidArg)+    , ("RdKafkaRespErrSsl", KafkaResponseError RdKafkaRespErrSsl)+    , ("RdKafkaRespErrUnknownProtocol", KafkaResponseError RdKafkaRespErrUnknownProtocol)+    , ("RdKafkaRespErrNotImplemented", KafkaResponseError RdKafkaRespErrNotImplemented)+    , ("RdKafkaRespErrAuthentication", KafkaResponseError RdKafkaRespErrAuthentication)+    , ("RdKafkaRespErrInconsistentGroupProtocol", KafkaResponseError RdKafkaRespErrInconsistentGroupProtocol)+    , ("RdKafkaRespErrTopicAuthorizationFailed", KafkaResponseError RdKafkaRespErrTopicAuthorizationFailed)+    , ("RdKafkaRespErrGroupAuthorizationFailed", KafkaResponseError RdKafkaRespErrGroupAuthorizationFailed)+    , ("RdKafkaRespErrClusterAuthorizationFailed", KafkaResponseError RdKafkaRespErrClusterAuthorizationFailed)+    , ("RdKafkaRespErrUnsupportedSaslMechanism", KafkaResponseError RdKafkaRespErrUnsupportedSaslMechanism)+    , ("RdKafkaRespErrIllegalSaslState", KafkaResponseError RdKafkaRespErrIllegalSaslState)+    , ("RdKafkaRespErrUnsupportedVersion", KafkaResponseError RdKafkaRespErrUnsupportedVersion)+    ]++nonFatalErrors :: [(String, KafkaError)]+nonFatalErrors =+    [ ("RdKafkaRespErrTimedOut", KafkaResponseError RdKafkaRespErrTimedOut)+    , ("RdKafkaRespErrPartitionEof", KafkaResponseError RdKafkaRespErrPartitionEof)+    , ("RdKafkaRespErrNoError", KafkaResponseError RdKafkaRespErrNoError)+    ]++isFatalTests :: [TestTree]+isFatalTests =+    [ testCase ("fatal: " <> label) (isFatal err @?= True)+    | (label, err) <- fatalErrors+    ]+        <> [ testCase ("non-fatal: " <> label) (isFatal err @?= False)+           | (label, err) <- nonFatalErrors+           ]++-------------------------------------------------------------------------------+-- isPollTimeout+-------------------------------------------------------------------------------++isPollTimeoutTests :: [TestTree]+isPollTimeoutTests =+    [ testCase "matches RdKafkaRespErrTimedOut" $+        isPollTimeout (KafkaResponseError RdKafkaRespErrTimedOut) @?= True+    , testCase "rejects RdKafkaRespErrPartitionEof" $+        isPollTimeout (KafkaResponseError RdKafkaRespErrPartitionEof) @?= False+    , testCase "rejects RdKafkaRespErrAuthentication" $+        isPollTimeout (KafkaResponseError RdKafkaRespErrAuthentication) @?= False+    , testCase "rejects KafkaBadConfiguration" $+        isPollTimeout KafkaBadConfiguration @?= False+    ]++-------------------------------------------------------------------------------+-- isPartitionEOF+-------------------------------------------------------------------------------++isPartitionEOFTests :: [TestTree]+isPartitionEOFTests =+    [ testCase "matches RdKafkaRespErrPartitionEof" $+        isPartitionEOF (KafkaResponseError RdKafkaRespErrPartitionEof) @?= True+    , testCase "rejects RdKafkaRespErrTimedOut" $+        isPartitionEOF (KafkaResponseError RdKafkaRespErrTimedOut) @?= False+    , testCase "rejects RdKafkaRespErrAuthentication" $+        isPartitionEOF (KafkaResponseError RdKafkaRespErrAuthentication) @?= False+    , testCase "rejects KafkaBadConfiguration" $+        isPartitionEOF KafkaBadConfiguration @?= False+    ]++-------------------------------------------------------------------------------+-- skipNonFatal / skipNonFatalExcept+-------------------------------------------------------------------------------++runStream ::+    (Stream IO a -> Stream IO b) ->+    [a] ->+    IO [b]+runStream f xs = Stream.fold Fold.toList (f (Stream.fromList xs))++timedOut, partitionEof, authentication :: KafkaError+timedOut = KafkaResponseError RdKafkaRespErrTimedOut+partitionEof = KafkaResponseError RdKafkaRespErrPartitionEof+authentication = KafkaResponseError RdKafkaRespErrAuthentication++skipNonFatalTests :: [TestTree]+skipNonFatalTests =+    [ testCase "drops poll-timeout, keeps authentication, keeps all Rights" dropsAndKeeps+    , testCase "passes all Rights through unchanged" passesAllRights+    , testCase "drops non-fatal Lefts even with empty extension list" dropsNonFatal+    , testCase "drops partition-EOF (non-fatal)" dropsPartitionEof+    ]+  where+    dropsAndKeeps :: Assertion+    dropsAndKeeps = do+        let input =+                [ Right (1 :: Int)+                , Left timedOut+                , Right 2+                , Left authentication+                , Right 3+                ]+        result <- runStream skipNonFatal input+        result @?= [Right 1, Right 2, Left authentication, Right 3]++    passesAllRights :: Assertion+    passesAllRights = do+        let input = [Right (n :: Int) | n <- [1 .. 5]]+        result <- runStream skipNonFatal input+        result @?= input++    dropsNonFatal :: Assertion+    dropsNonFatal = do+        let input :: [Either KafkaError Int]+            input =+                [ Left timedOut+                , Left partitionEof+                , Left (KafkaResponseError RdKafkaRespErrNoError)+                ]+        result <- runStream skipNonFatal input+        result @?= []++    dropsPartitionEof :: Assertion+    dropsPartitionEof = do+        let input =+                [ Right (1 :: Int)+                , Left partitionEof+                , Right 2+                ]+        result <- runStream skipNonFatal input+        result @?= [Right 1, Right 2]++skipNonFatalExceptTests :: [TestTree]+skipNonFatalExceptTests =+    [ testCase "keeps poll-timeout when isPollTimeout is in the list" keepsTimeouts+    , testCase "keeps partition-EOF when isPartitionEOF is in the list" keepsPartitionEof+    , testCase "empty predicate list behaves like skipNonFatal" emptyListEquivalent+    ]+  where+    keepsTimeouts :: Assertion+    keepsTimeouts = do+        let input =+                [ Right (1 :: Int)+                , Left timedOut+                , Right 2+                , Left authentication+                , Right 3+                ]+        result <- runStream (skipNonFatalExcept [isPollTimeout]) input+        result @?= [Right 1, Left timedOut, Right 2, Left authentication, Right 3]++    keepsPartitionEof :: Assertion+    keepsPartitionEof = do+        let input =+                [ Right (1 :: Int)+                , Left partitionEof+                , Left timedOut+                , Right 2+                ]+        result <- runStream (skipNonFatalExcept [isPartitionEOF]) input+        result @?= [Right 1, Left partitionEof, Right 2]++    emptyListEquivalent :: Assertion+    emptyListEquivalent = do+        let input =+                [ Right (1 :: Int)+                , Left timedOut+                , Left authentication+                , Right 2+                ]+        viaExcept <- runStream (skipNonFatalExcept []) input+        viaPlain <- runStream skipNonFatal input+        viaExcept @?= viaPlain

@@ -1,7 +1,7 @@ module Main (main) where  import Kafka.Streamly.CombinatorsTest qualified as CombinatorsTest-import Kafka.Streamly.SourceTest qualified as SourceTest+import Kafka.Streamly.StreamTest qualified as StreamTest import Test.Tasty (TestTree, defaultMain, testGroup)  main :: IO ()@@ -11,6 +11,6 @@ tests =     testGroup         "hw-kafka-streamly"-        [ SourceTest.tests+        [ StreamTest.tests         , CombinatorsTest.tests         ]