diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 --- /dev/null +++ b/CHANGELOG.md @@ -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`. diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -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. diff --git a/README.md b/README.md new file mode 100644 --- /dev/null +++ b/README.md @@ -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 diff --git a/hw-kafka-streamly.cabal b/hw-kafka-streamly.cabal new file mode 100644 --- /dev/null +++ b/hw-kafka-streamly.cabal @@ -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 diff --git a/src/Kafka/Streamly/Combinators.hs b/src/Kafka/Streamly/Combinators.hs new file mode 100644 --- /dev/null +++ b/src/Kafka/Streamly/Combinators.hs @@ -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) diff --git a/src/Kafka/Streamly/Sink.hs b/src/Kafka/Streamly/Sink.hs new file mode 100644 --- /dev/null +++ b/src/Kafka/Streamly/Sink.hs @@ -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) diff --git a/src/Kafka/Streamly/Source.hs b/src/Kafka/Streamly/Source.hs new file mode 100644 --- /dev/null +++ b/src/Kafka/Streamly/Source.hs @@ -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 #-} diff --git a/test/Kafka/Streamly/CombinatorsTest.hs b/test/Kafka/Streamly/CombinatorsTest.hs new file mode 100644 --- /dev/null +++ b/test/Kafka/Streamly/CombinatorsTest.hs @@ -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 @?= [] + ] diff --git a/test/Kafka/Streamly/SourceTest.hs b/test/Kafka/Streamly/SourceTest.hs new file mode 100644 --- /dev/null +++ b/test/Kafka/Streamly/SourceTest.hs @@ -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 diff --git a/test/Main.hs b/test/Main.hs new file mode 100644 --- /dev/null +++ b/test/Main.hs @@ -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 + ]