packages feed

hw-kafka-streamly (empty) → 0.1.0.0

raw patch · 10 files changed

+1324/−0 lines, 10 filesdep +basedep +bytestringdep +exceptions

Dependencies added: base, bytestring, exceptions, hw-kafka-client, hw-kafka-streamly, streamly-core, tasty, tasty-hunit, tasty-quickcheck

Files

+ CHANGELOG.md view
@@ -0,0 +1,51 @@+# Changelog++All notable changes to this project will be documented in this file.++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)++Initial release.++### Added++- `Kafka.Streamly.Source` — three consumer stream variants (`kafkaSource`,+  `kafkaSourceAutoClose`, `kafkaSourceNoClose`) for different resource-ownership+  models; error predicates (`isFatal`, `isPollTimeout`, `isPartitionEOF`);+  error filters (`skipNonFatal`, `skipNonFatalExcept`); and value-mapping+  helpers built on `Bifunctor` (`mapFirst`, `mapValue`, `bimapValue`).+- `Kafka.Streamly.Sink` — two producer fold variants (`kafkaSink` for+  per-record sends, `kafkaBatchSink` for `[ProducerRecord]` batches) and a+  `withKafkaProducer` bracket helper that flushes and closes on exit.+- `Kafka.Streamly.Combinators` — `batchByOrFlush` and `batchByOrFlushEither`+  for size-bounded batching with explicit flush, and `throwLeft` /+  `throwLeftSatisfy` for raising error values as exceptions.+- Module-level Haddock with worked examples and `@since 0.1.0.0` tags on+  every exported identifier.++### Fixed++- `withKafkaProducer` no longer flushes twice on teardown+  (`Kafka.Producer.closeProducer` already flushes internally).++### Changed++- `batchByOrFlush` and `batchByOrFlushEither` now reject non-positive+  `BatchSize` with an explicit error rather than silently emitting+  singleton batches.+- Value-mapping helpers pruned: the nine `sequenceValue*`, `traverseValue*`,+  and `bitraverseValue*`/`bisequenceValue` variants have been removed.+  Callers who relied on them can use `fmap`/`bimap`/`traverse`/`bitraverse`+  inline. `mapFirst`, `mapValue`, and `bimapValue` remain, re-documented as+  `Bifunctor`/`Functor` lifts.++### Tests++- Initial pure test suite covering error predicates (`isFatal`,+  `isPollTimeout`, `isPartitionEOF`), error filters (`skipNonFatal`,+  `skipNonFatalExcept`), batching combinators (`batchByOrFlush`,+  `batchByOrFlushEither`), and exception combinators (`throwLeft`,+  `throwLeftSatisfy`). Uses `tasty` + `tasty-hunit` + `tasty-quickcheck`.+  Run with `cabal test`.
+ LICENSE view
@@ -0,0 +1,21 @@+MIT License++Copyright (c) 2026 Nadeem Bitar++Permission is hereby granted, free of charge, to any person obtaining a copy+of this software and associated documentation files (the "Software"), to deal+in the Software without restriction, including without limitation the rights+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell+copies of the Software, and to permit persons to whom the Software is+furnished to do so, subject to the following conditions:++The above copyright notice and this permission notice shall be included in all+copies or substantial portions of the Software.++THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE+SOFTWARE.
+ README.md view
@@ -0,0 +1,115 @@+# hw-kafka-streamly++Streamly streaming integration for [`hw-kafka-client`][hw-kafka-client] — the+Haskell binding to Apache Kafka via [librdkafka][librdkafka].++`hw-kafka-streamly` exposes Kafka consumers as composable Streamly `Stream`s+and Kafka producers as Streamly `Fold`s, so message processing can be expressed+in the same vocabulary used elsewhere in a Streamly pipeline. Resource handling+(consumer/producer creation, close, flush) is layered so the caller picks the+amount of bracketing they want.++## Installation++Add to your `.cabal` file:++```cabal+build-depends:+  , hw-kafka-streamly  >=0.1 && <0.2+  , hw-kafka-client    >=5.3 && <6+  , streamly-core      >=0.4 && <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).++## Modules++- `Kafka.Streamly.Source` — consumer streams, error predicates and filters,+  value-mapping helpers built on `Bifunctor`/`Bitraversable`.+- `Kafka.Streamly.Sink` — 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 underlying `KafkaConsumer`:++- `kafkaSource` — 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+  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+  it open. Use this when the consumer outlives the stream.++```haskell+import Kafka.Consumer+import Kafka.Streamly.Source (kafkaSource, skipNonFatal)+import Streamly.Data.Stream qualified as Stream++main :: IO ()+main = do+  let props = brokersList ["localhost:9092"]+           <> groupId "my-group"+           <> noAutoCommit+      sub  = topics ["events"] <> offsetReset Earliest+  Stream.fold (Fold.drainBy print)+    . skipNonFatal+    $ kafkaSource props sub (Timeout 1000)+```++## Producing++Producer sinks are Streamly `Fold`s:++```haskell+import Kafka.Producer+import Kafka.Streamly.Sink (kafkaSink, withKafkaProducer)+import Streamly.Data.Fold qualified as Fold+import Streamly.Data.Stream qualified as Stream++main :: IO ()+main = do+  let props = brokersList ["localhost:9092"]+  result <- withKafkaProducer props $ \producer ->+    Stream.fold (kafkaSink producer)+      . fmap mkRecord+      $ Stream.fromList ["a", "b", "c"]+  print result+  where+    mkRecord v =+      ProducerRecord+        { prTopic     = TopicName "events"+        , prPartition = UnassignedPartition+        , prKey       = Nothing+        , prValue     = Just v+        , prHeaders   = mempty+        }+```++## Cookbook++For end-to-end runnable examples — concurrent consumers, batching producers,+error handling, transform pipelines — see the companion+[`hw-kafka-streamly-jitsurei`][jitsurei] package in this repository. It is not+published to Hackage; clone the repo and run the executables locally.++## Design notes++The original design plan for the bindings lives at+`docs/plans/1-streamly-bindings-for-hw-kafka-client.md`.++## License++MIT — see [LICENSE](./LICENSE).++[hw-kafka-client]: https://hackage.haskell.org/package/hw-kafka-client+[librdkafka]: https://github.com/confluentinc/librdkafka+[jitsurei]: ../hw-kafka-streamly-jitsurei
+ hw-kafka-streamly.cabal view
@@ -0,0 +1,90 @@+cabal-version:   3.4+name:            hw-kafka-streamly+version:         0.1.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.++license:         MIT+license-file:    LICENSE+author:          Nadeem Bitar+maintainer:      Nadeem Bitar+copyright:       2026 Nadeem Bitar+category:        Kafka+homepage:        https://github.com/shinzui/hw-kafka-streamly+bug-reports:     https://github.com/shinzui/hw-kafka-streamly/issues+build-type:      Simple+tested-with:     GHC ==9.12.2+extra-doc-files:+  CHANGELOG.md+  README.md++source-repository head+  type:     git+  location: https://github.com/shinzui/hw-kafka-streamly.git++common warnings+  ghc-options:+    -Wall -Wcompat -Widentities -Wincomplete-uni-patterns+    -Wincomplete-record-updates -Wredundant-constraints+    -fhide-source-paths -Wmissing-export-lists -Wpartial-fields+    -Wmissing-deriving-strategies++library+  import:             warnings+  exposed-modules:+    Kafka.Streamly.Combinators+    Kafka.Streamly.Sink+    Kafka.Streamly.Source++  build-depends:+    , base             >=4.21 && <5+    , bytestring       >=0.11 && <0.13+    , exceptions       >=0.10 && <1+    , hw-kafka-client  >=5.3  && <6+    , streamly-core    >=0.3  && <0.5++  hs-source-dirs:     src+  default-language:   GHC2024+  default-extensions:+    DataKinds+    DuplicateRecordFields+    ImportQualifiedPost+    LambdaCase+    NoFieldSelectors+    OverloadedRecordDot+    OverloadedStrings++test-suite hw-kafka-streamly-test+  import:             warnings+  type:               exitcode-stdio-1.0+  main-is:            Main.hs+  hs-source-dirs:     test+  other-modules:+    Kafka.Streamly.CombinatorsTest+    Kafka.Streamly.SourceTest++  build-depends:+    , base               >=4.21 && <5+    , exceptions         >=0.10 && <1+    , hw-kafka-client    >=5.3  && <6+    , hw-kafka-streamly+    , streamly-core      >=0.3  && <0.5+    , tasty              >=1.4  && <2+    , tasty-hunit        >=0.10 && <0.11+    , tasty-quickcheck   >=0.10 && <0.12++  default-language:   GHC2024+  default-extensions:+    DataKinds+    DuplicateRecordFields+    ImportQualifiedPost+    LambdaCase+    NoFieldSelectors+    OverloadedRecordDot+    OverloadedStrings++  ghc-options:        -threaded -rtsopts -with-rtsopts=-N
+ src/Kafka/Streamly/Combinators.hs view
@@ -0,0 +1,136 @@+{- |+Module      : Kafka.Streamly.Combinators+Description : Auxiliary combinators for Streamly–Kafka pipelines.++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.++Re-exports t'Kafka.Types.BatchSize' from "Kafka.Types" for convenience.+-}+module Kafka.Streamly.Combinators (+    -- * Types+    BatchSize (..),++    -- * Batching combinators+    batchByOrFlush,+    batchByOrFlushEither,++    -- * Error handling+    throwLeft,+    throwLeftSatisfy,+) where++import Control.Monad.Catch (Exception, MonadThrow, throwM)+import Kafka.Types (BatchSize (..))+import Streamly.Data.Scanl qualified as Scanl+import Streamly.Data.Stream (Stream)+import Streamly.Data.Stream qualified as Stream++{- | Throws the left part of a value as an exception, passing right values+through.++@since 0.1.0.0+-}+throwLeft ::+    (MonadThrow m, Exception e) =>+    Stream m (Either e a) ->+    Stream m a+throwLeft = Stream.mapMaybeM $ \case+    Left e -> throwM e+    Right a -> pure (Just a)+{-# INLINE throwLeft #-}++{- | Throws the left part of a value as an exception if it satisfies the+predicate. Non-matching left values and all right values pass through+unchanged.++@since 0.1.0.0+-}+throwLeftSatisfy ::+    (MonadThrow m, Exception e) =>+    (e -> Bool) ->+    Stream m (Either e a) ->+    Stream m (Either e a)+throwLeftSatisfy p = Stream.mapMaybeM $ \case+    Left e | p e -> throwM e+    v -> pure (Just v)+{-# INLINE throwLeftSatisfy #-}++{- | Batch stream elements by size, flushing on 'Nothing'.++Elements wrapped in 'Just' accumulate into a batch. When the batch reaches+the specified size or a 'Nothing' is received, the current batch is emitted.+Empty batches are not emitted. A final incomplete batch is emitted when the+stream ends.++Raises an 'error' if the t'BatchSize' is non-positive — @BatchSize 0@ would+emit a singleton batch for every element, which is never what the caller+wants.++@since 0.1.0.0+-}+batchByOrFlush ::+    (Monad m) =>+    BatchSize ->+    Stream m (Maybe a) ->+    Stream m [a]+batchByOrFlush n input =+    batchInternal "batchByOrFlush" n (Stream.append input (Stream.fromPure Nothing))+{-# INLINE batchByOrFlush #-}++{- | Batch stream elements by size, flushing on 'Left'.++'Right' values accumulate into a batch. When the batch reaches+the specified size or a 'Left' is received, the current batch is emitted.+Empty batches are not emitted. A final incomplete batch is emitted when the+stream ends.++Raises an 'error' if the t'BatchSize' is non-positive.++@since 0.1.0.0+-}+batchByOrFlushEither ::+    (Monad m) =>+    BatchSize ->+    Stream m (Either e a) ->+    Stream m [a]+batchByOrFlushEither n input =+    batchInternal "batchByOrFlushEither" n $+        Stream.append (fmap eitherToMaybe input) (Stream.fromPure Nothing)+  where+    eitherToMaybe (Left _) = Nothing+    eitherToMaybe (Right a) = Just a+{-# INLINE batchByOrFlushEither #-}++-- Internal batching implementation shared by both combinators.+batchInternal ::+    (Monad m) =>+    String ->+    BatchSize ->+    Stream m (Maybe a) ->+    Stream m [a]+batchInternal caller (BatchSize n)+    | n <= 0 = error (caller <> ": BatchSize must be positive, got " <> show n)+    | otherwise =+        Stream.catMaybes+            . fmap extract+            . Stream.scanl (Scanl.mkScanl step initial)+  where+    initial :: (Int, [a], Maybe [a])+    initial = (0, [], Nothing)++    extract :: (Int, [a], Maybe [a]) -> Maybe [a]+    extract (_, _, out) = out++    step :: (Int, [a], Maybe [a]) -> Maybe a -> (Int, [a], Maybe [a])+    step (_, acc, _) Nothing =+        if null acc+            then (0, [], Nothing)+            else (0, [], Just (reverse acc))+    step (i, acc, _) (Just a) =+        let acc' = a : acc+         in if i + 1 >= n+                then (0, [], Just (reverse acc'))+                else (i + 1, acc', Nothing)
+ src/Kafka/Streamly/Sink.hs view
@@ -0,0 +1,146 @@+{- |+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)
+ src/Kafka/Streamly/Source.hs view
@@ -0,0 +1,329 @@+{- |+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 #-}
+ test/Kafka/Streamly/CombinatorsTest.hs view
@@ -0,0 +1,216 @@+module Kafka.Streamly.CombinatorsTest (tests) where++import Control.Exception (+    ErrorCall,+    Exception,+    SomeException,+    try,+ )+import Data.Either (rights)+import Data.Maybe (catMaybes)+import Kafka.Streamly.Combinators (+    BatchSize (..),+    batchByOrFlush,+    batchByOrFlushEither,+    throwLeft,+    throwLeftSatisfy,+ )+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, assertFailure, testCase, (@?=))+import Test.Tasty.QuickCheck (Positive (..), Property, ioProperty, testProperty, (===))++tests :: TestTree+tests =+    testGroup+        "Combinators"+        [ testGroup "batchByOrFlush unit" batchByOrFlushUnitTests+        , testGroup "batchByOrFlush property" batchByOrFlushProperties+        , testGroup "batchByOrFlushEither unit" batchByOrFlushEitherUnitTests+        , testGroup "batchByOrFlushEither property" batchByOrFlushEitherProperties+        , testGroup "BatchSize validation" batchSizeValidationTests+        , testGroup "throwLeft" throwLeftTests+        , testGroup "throwLeftSatisfy" throwLeftSatisfyTests+        ]++-------------------------------------------------------------------------------+-- Helpers+-------------------------------------------------------------------------------++runBatchM :: BatchSize -> [Maybe Int] -> IO [[Int]]+runBatchM n xs =+    Stream.fold Fold.toList (batchByOrFlush n (Stream.fromList xs))++runBatchE :: BatchSize -> [Either () Char] -> IO [String]+runBatchE n xs =+    Stream.fold Fold.toList (batchByOrFlushEither n (Stream.fromList xs))++runStream :: (Stream IO a -> Stream IO b) -> [a] -> IO [b]+runStream f xs = Stream.fold Fold.toList (f (Stream.fromList xs))++-------------------------------------------------------------------------------+-- batchByOrFlush unit+-------------------------------------------------------------------------------++batchByOrFlushUnitTests :: [TestTree]+batchByOrFlushUnitTests =+    [ testCase "groups into full batch and trailing partial" $ do+        result <- runBatchM (BatchSize 3) [Just 1, Just 2, Just 3, Just 4, Just 5]+        result @?= [[1, 2, 3], [4, 5]]+    , testCase "Nothing flushes the current partial batch" $ do+        result <- runBatchM (BatchSize 3) [Just 1, Nothing, Just 2, Just 3]+        result @?= [[1], [2, 3]]+    , testCase "empty input produces no batches" $ do+        result <- runBatchM (BatchSize 3) []+        result @?= []+    , testCase "only-Nothing input produces no empty batches" $ do+        result <- runBatchM (BatchSize 3) [Nothing, Nothing, Nothing]+        result @?= []+    , testCase "exact fill then trailing partial" $ do+        result <- runBatchM (BatchSize 2) [Just 1, Just 2, Just 3]+        result @?= [[1, 2], [3]]+    , testCase "BatchSize 1 emits singletons" $ do+        result <- runBatchM (BatchSize 1) [Just 1, Just 2, Just 3]+        result @?= [[1], [2], [3]]+    ]++-------------------------------------------------------------------------------+-- batchByOrFlush QuickCheck properties+-------------------------------------------------------------------------------++batchByOrFlushProperties :: [TestTree]+batchByOrFlushProperties =+    [ testProperty "concat of batches equals catMaybes of input" prop_concatEqualsCatMaybes+    , testProperty "every emitted batch has 1 <= length <= BatchSize" prop_batchLengths+    ]+  where+    prop_concatEqualsCatMaybes :: Positive Int -> [Maybe Int] -> Property+    prop_concatEqualsCatMaybes (Positive n) xs = ioProperty $ do+        result <- runBatchM (BatchSize n) xs+        pure (concat result === catMaybes xs)++    prop_batchLengths :: Positive Int -> [Maybe Int] -> Property+    prop_batchLengths (Positive n) xs = ioProperty $ do+        result <- runBatchM (BatchSize n) xs+        pure $+            all (\b -> not (null b) && length b <= n) result+                === True++-------------------------------------------------------------------------------+-- batchByOrFlushEither unit+-------------------------------------------------------------------------------++batchByOrFlushEitherUnitTests :: [TestTree]+batchByOrFlushEitherUnitTests =+    [ testCase "Left flushes, Right fills the batch" $ do+        result <- runBatchE (BatchSize 2) [Right 'a', Right 'b', Left (), Right 'c']+        result @?= ["ab", "c"]+    , testCase "leading Left produces no empty batch" $ do+        result <- runBatchE (BatchSize 2) [Left (), Right 'a']+        result @?= ["a"]+    , testCase "empty input produces no batches" $ do+        result <- runBatchE (BatchSize 2) []+        result @?= []+    ]++-------------------------------------------------------------------------------+-- batchByOrFlushEither QuickCheck properties+-------------------------------------------------------------------------------++batchByOrFlushEitherProperties :: [TestTree]+batchByOrFlushEitherProperties =+    [ testProperty "concat of batches equals rights of input" prop_concatEqualsRights+    , testProperty "every emitted batch has 1 <= length <= BatchSize" prop_batchLengthsE+    ]+  where+    prop_concatEqualsRights :: Positive Int -> [Either () Int] -> Property+    prop_concatEqualsRights (Positive n) xs = ioProperty $ do+        result <- Stream.fold Fold.toList (batchByOrFlushEither (BatchSize n) (Stream.fromList xs))+        pure (concat result === rights xs)++    prop_batchLengthsE :: Positive Int -> [Either () Int] -> Property+    prop_batchLengthsE (Positive n) xs = ioProperty $ do+        result <- Stream.fold Fold.toList (batchByOrFlushEither (BatchSize n) (Stream.fromList xs))+        pure $+            all (\b -> not (null b) && length b <= n) result+                === True++-------------------------------------------------------------------------------+-- BatchSize validation+-------------------------------------------------------------------------------++batchSizeValidationTests :: [TestTree]+batchSizeValidationTests =+    [ testCase "batchByOrFlush errors on BatchSize 0" $+        assertErrors+            (Stream.fold Fold.toList (batchByOrFlush (BatchSize 0) (Stream.fromList [Just (1 :: Int)])))+    , testCase "batchByOrFlush errors on negative BatchSize" $+        assertErrors+            (Stream.fold Fold.toList (batchByOrFlush (BatchSize (-3)) (Stream.fromList [Just (1 :: Int)])))+    , testCase "batchByOrFlushEither errors on BatchSize 0" $+        assertErrors+            (Stream.fold Fold.toList (batchByOrFlushEither (BatchSize 0) (Stream.fromList [Right 'a' :: Either () Char])))+    ]+  where+    assertErrors :: (Show a) => IO a -> Assertion+    assertErrors action = do+        result <- try @ErrorCall action+        case result of+            Left _ -> pure ()+            Right v -> assertFailure $ "expected ErrorCall, got: " <> show v++-------------------------------------------------------------------------------+-- throwLeft / throwLeftSatisfy+-------------------------------------------------------------------------------++data TestErr = Good | Bad+    deriving stock (Eq, Show)++instance Exception TestErr++throwLeftTests :: [TestTree]+throwLeftTests =+    [ testCase "passes Rights through unchanged" $ do+        let input = [Right 1, Right 2, Right 3] :: [Either TestErr Int]+        result <- runStream throwLeft input+        result @?= [1, 2, 3]+    , testCase "throws on Left" $ do+        let input = [Right 1, Left Bad, Right 3] :: [Either TestErr Int]+        caught <- try @TestErr (runStream throwLeft input)+        caught @?= Left Bad+    , testCase "empty stream produces no exception" $ do+        let input = [] :: [Either TestErr Int]+        result <- runStream throwLeft input+        result @?= []+    ]++throwLeftSatisfyTests :: [TestTree]+throwLeftSatisfyTests =+    [ testCase "passes non-matching Lefts and all Rights through" $ do+        let input =+                [ Right 1+                , Left Good+                , Right 2+                ] ::+                    [Either TestErr Int]+        result <- runStream (throwLeftSatisfy (== Bad)) input+        result @?= input+    , testCase "throws only on matching Left" $ do+        let input =+                [ Right 1+                , Left Good+                , Right 2+                , Left Bad+                ] ::+                    [Either TestErr Int]+        caught <- try @SomeException (runStream (throwLeftSatisfy (== Bad)) input)+        case caught of+            Left _ -> pure ()+            Right v -> assertFailure $ "expected exception, got: " <> show v+    , testCase "empty stream produces no exception" $ do+        let input = [] :: [Either TestErr Int]+        result <- runStream (throwLeftSatisfy (== Bad)) input+        result @?= []+    ]
+ test/Kafka/Streamly/SourceTest.hs view
@@ -0,0 +1,204 @@+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
+ test/Main.hs view
@@ -0,0 +1,16 @@+module Main (main) where++import Kafka.Streamly.CombinatorsTest qualified as CombinatorsTest+import Kafka.Streamly.SourceTest qualified as SourceTest+import Test.Tasty (TestTree, defaultMain, testGroup)++main :: IO ()+main = defaultMain tests++tests :: TestTree+tests =+    testGroup+        "hw-kafka-streamly"+        [ SourceTest.tests+        , CombinatorsTest.tests+        ]