freckle-kafka 0.0.0.1 → 0.0.0.2

raw patch · 5 files changed

+74/−12 lines, 5 filesdep ~BlammoPVP: minor bump suggested

API additions: PVP suggests at least a minor version bump

Dependency ranges changed: Blammo

API changes (from Hackage documentation)

Files

• CHANGELOG.md +5/−1
• README.md +1/−1
• freckle-kafka.cabal +2/−2
• library/Freckle/App/Kafka/Consumer.hs +64/−6
• package.yaml +2/−2

@@ -1,4 +1,8 @@-## [_Unreleased_](https://github.com/freckle/freckle-app/compare/freckle-kafka-v0.0.0.1...main)+## [_Unreleased_](freckle-kafka-v0.0.0.2...main)++## [v1.21.1.0](https://github.com/freckle/freckle-app/compare/freckle-kafka-v0.0.0.1...freckle-kafka-v0.0.0.2)++Add `Freckle.App.Kafka.Consumer.runConsumerBatched`  ## [v0.0.0.1](https://github.com/freckle/freckle-app/compare/freckle-kafka-v0.0.0.0...freckle-kafka-v0.0.0.1) 

@@ -1,6 +1,6 @@ # freckle-kafka -Some extensions to the `hw-kafka-client` library.+Some extensions to the [hw-kafka-client](https://hackage.haskell.org/package/hw-kafka-client) library.  --- 

@@ -1,6 +1,6 @@ cabal-version:      1.18 name:               freckle-kafka-version:            0.0.0.1+version:            0.0.0.2 license:            MIT license-file:       LICENSE maintainer:         Freckle Education@@ -42,7 +42,7 @@         -Wno-safe -Wno-unsafe      build-depends:-        Blammo >=2.0.0.0,+        Blammo >=2.1.0.0,         aeson >=2.0.3.0,         annotated-exception >=0.2.0.4,         base >=4.16.4.0 && <5,

@@ -6,6 +6,7 @@   , KafkaConsumerConfig (..)   , envKafkaConsumerConfig   , runConsumer+  , runConsumerBatched   ) where  import Prelude@@ -18,17 +19,19 @@   , displayException   ) import Control.Exception.Annotated.UnliftIO qualified as Annotated-import Control.Lens (Lens', view)+import Control.Lens (Lens', over, view, _Left) import Control.Monad (forever, (<=<)) import Control.Monad.IO.Class (MonadIO) import Control.Monad.Reader (MonadReader) import Data.Aeson import Data.ByteString (ByteString)-import Data.Foldable (for_)-import Data.List.NonEmpty (NonEmpty)+import Data.Either (partitionEithers)+import Data.Foldable (for_, traverse_)+import Data.List.NonEmpty (NonEmpty, nonEmpty) import Data.List.NonEmpty qualified as NE import Data.Map.Strict (Map) import Data.Map.Strict qualified as Map+import Data.Maybe (mapMaybe) import Data.Text (Text) import Data.Text qualified as T import Data.Text.Encoding qualified as T@@ -198,7 +201,7 @@         "kafka.consumer"         (defaultSpanArguments {Trace.kind = Consumer})       $ do-        mRecord <- fromKafkaError =<< pollMessage consumer kTimeout+        mRecord <- fromKafkaError =<< pollMessage consumer (kafkaTimeout pollTimeout)          for_ (crValue =<< mRecord) $ \bs -> do           a <-@@ -207,8 +210,6 @@                 eitherDecodeStrict bs           inSpan "kafka.consumer.message.handle" defaultSpanArguments $ onMessage a  where-  kTimeout = Kafka.Timeout $ timeoutMs pollTimeout-   handlers =     [ Annotated.Handler $         logErrorNS "kafka"@@ -219,6 +220,63 @@           . annotatedExceptionMessageFrom @KafkaMessageDecodeError             (const "Could not decode message value")     ]++data BatchConsumerError+  = BatchConsumerDecodeError KafkaMessageDecodeError+  | BatchConsumerKafkaError KafkaError+  deriving stock (Show)++runConsumerBatched+  :: forall a m env+   . ( MonadUnliftIO m+     , MonadReader env m+     , MonadLogger m+     , MonadTracer m+     , HasKafkaConsumer env+     , FromJSON a+     , HasCallStack+     )+  => Timeout+  -> BatchSize+  -> ([a] -> m ())+  -> m ()+runConsumerBatched pollTimeout batchSize onBatch =+  forever $ do+    consumer <- view kafkaConsumerL++    errors <- inSpan+      "kafka.batchConsumer"+      (defaultSpanArguments {Trace.kind = Consumer})+      $ do+        (errors, records) <- do+          (kafkaErrors, batch) <-+            partitionEithers+              <$> pollMessageBatch consumer (kafkaTimeout pollTimeout) batchSize+          (decodeErrors, records) <- fmap partitionEithers $ inSpan "kafka.batchConsumer.messages.decode" defaultSpanArguments $ do+            let errorDecode bs = over _Left (KafkaMessageDecodeError bs) $ eitherDecodeStrict @a bs+            pure $ mapMaybe (fmap errorDecode . crValue) batch+          pure+            ( fmap BatchConsumerKafkaError kafkaErrors+                <> fmap BatchConsumerDecodeError decodeErrors+            , records+            )++        inSpan "kafka.batchConsumer.messages.handle" defaultSpanArguments $+          onBatch records+        pure $ nonEmpty errors++    traverse_ logErrors errors+ where+  displayConsumerError = \case+    BatchConsumerDecodeError e -> displayException e+    BatchConsumerKafkaError e -> displayException e+  logErrors errors =+    logErrorNS "kafka" $+      "Batch consumer errors"+        :# ["errors" .= fmap displayConsumerError errors]++kafkaTimeout :: Timeout -> Kafka.Timeout+kafkaTimeout = Kafka.Timeout . timeoutMs  -- | Like 'annotatedExceptionMessage', but use the supplied function to --   construct an initial 'Message' that it will augment.

@@ -1,5 +1,5 @@ name: freckle-kafka-version: 0.0.0.1+version: 0.0.0.2 maintainer: Freckle Education category: Database github: freckle/freckle-app@@ -54,7 +54,7 @@ library:   source-dirs: library   dependencies:-    - Blammo >= 2.0.0.0+    - Blammo     - aeson     - annotated-exception     - bytestring