freckle-app 1.9.2.0 → 1.9.2.1
raw patch · 4 files changed
+26/−18 lines, 4 files
Files
- CHANGELOG.md +4/−0
- freckle-app.cabal +1/−1
- library/Freckle/App/Kafka/Consumer.hs +20/−16
- package.yaml +1/−1
CHANGELOG.md view
@@ -1,5 +1,9 @@ ## [_Unreleased_](https://github.com/freckle/freckle-app/compare/v1.9.1.1...main) +## [v1.9.2.1](https://github.com/freckle/freckle-app/compare/v1.9.2.0...v1.9.2.1)++- Fixed segfault bug in `Freckle.App.Kafka.Consumer`.+ ## [v1.9.2.0](https://github.com/freckle/freckle-app/compare/v1.9.1.1...v1.9.2.0) - Add `Freckle.App.Kafka.Consumer` for consuming Kafka events.
freckle-app.cabal view
@@ -1,6 +1,6 @@ cabal-version: 1.18 name: freckle-app-version: 1.9.2.0+version: 1.9.2.1 license: MIT license-file: LICENSE maintainer: Freckle Education
library/Freckle/App/Kafka/Consumer.hs view
@@ -11,8 +11,10 @@ import Freckle.App.Prelude import Blammo.Logging+import Control.Concurrent import qualified Control.Immortal as Immortal import Control.Lens (Lens', view)+import Control.Monad (forever) import Data.Aeson import qualified Data.List.NonEmpty as NE import qualified Data.Map.Strict as Map@@ -142,22 +144,24 @@ => Timeout -> (a -> m ()) -> m ()-runConsumer pollTimeout onMessage = void $ Immortal.create $ \thread -> Immortal.onUnexpectedFinish thread handleException $ do- consumer <- view kafkaConsumerL- eMessage <-- pollMessage consumer $ Kafka.Timeout $ timeoutMs pollTimeout- case eMessage of- Left (KafkaResponseError RdKafkaRespErrTimedOut) -> logDebug $ "Polling timeout"- Left err -> logError $ "Error polling for message from Kafka" :# ["error" .= show err]- Right m@(ConsumerRecord {..}) -> for_ crValue $ \bs -> do- case eitherDecodeStrict bs of- Left err -> logError $ "Could not decode message value" :# ["error" .= err]- Right a -> do- onMessage a- maybe- (logInfo "Committed offsets")- (\err -> logError $ "Error committing offsets" :# ["error" .= show err])- =<< commitOffsetMessage OffsetCommit consumer m+runConsumer pollTimeout onMessage = do+ void $ Immortal.create $ \thread -> Immortal.onUnexpectedFinish thread handleException $ do+ consumer <- view kafkaConsumerL+ eMessage <-+ pollMessage consumer $ Kafka.Timeout $ timeoutMs pollTimeout+ case eMessage of+ Left (KafkaResponseError RdKafkaRespErrTimedOut) -> logDebug $ "Polling timeout"+ Left err -> logError $ "Error polling for message from Kafka" :# ["error" .= show err]+ Right m@(ConsumerRecord {..}) -> for_ crValue $ \bs -> do+ case eitherDecodeStrict bs of+ Left err -> logError $ "Could not decode message value" :# ["error" .= err]+ Right a -> do+ onMessage a+ maybe+ (logInfo "Committed offsets")+ (\err -> logError $ "Error committing offsets" :# ["error" .= show err])+ =<< commitOffsetMessage OffsetCommit consumer m+ forever $ liftIO $ threadDelay maxBound where handleException = \case Left ex ->
package.yaml view
@@ -1,5 +1,5 @@ name: freckle-app-version: 1.9.2.0+version: 1.9.2.1 maintainer: Freckle Education category: Utils github: freckle/freckle-app