freckle-kafka 0.0.0.0 → 0.0.0.1
raw patch · 5 files changed
+64/−37 lines, 5 filesdep +bytestringdep +mtldep −reludedep ~freckle-envPVP ok
version bump matches the API change (PVP)
Dependencies added: bytestring, mtl
Dependencies removed: relude
Dependency ranges changed: freckle-env
API changes (from Hackage documentation)
Files
- CHANGELOG.md +5/−1
- freckle-kafka.cabal +4/−3
- library/Freckle/App/Kafka/Consumer.hs +33/−18
- library/Freckle/App/Kafka/Producer.hs +19/−13
- package.yaml +3/−2
CHANGELOG.md view
@@ -1,4 +1,8 @@-## [_Unreleased_](https://github.com/freckle/freckle-app/compare/freckle-kafka-v0.0.0.0...main)+## [_Unreleased_](https://github.com/freckle/freckle-app/compare/freckle-kafka-v0.0.0.1...main)++## [v0.0.0.1](https://github.com/freckle/freckle-app/compare/freckle-kafka-v0.0.0.0...freckle-kafka-v0.0.0.1)++Drop `relude` dependency ## [v0.0.0.0](https://github.com/freckle/freckle-app/tree/freckle-kafka-v0.0.0.0/freckle-kafka)
freckle-kafka.cabal view
@@ -1,6 +1,6 @@ cabal-version: 1.18 name: freckle-kafka-version: 0.0.0.0+version: 0.0.0.1 license: MIT license-file: LICENSE maintainer: Freckle Education@@ -46,12 +46,13 @@ aeson >=2.0.3.0, annotated-exception >=0.2.0.4, base >=4.16.4.0 && <5,+ bytestring >=0.11.4.0, containers >=0.6.5.1,- freckle-env >=0.0.0.0,+ freckle-env >=0.0.1.1, hs-opentelemetry-sdk >=0.0.3.6, hw-kafka-client >=5.0.0, lens >=5.1.1,- relude >=1.1.0.0,+ mtl >=2.2.2, resource-pool >=0.4.0.0, text >=1.2.5.0, time >=1.11.1.1,
library/Freckle/App/Kafka/Consumer.hs view
@@ -8,19 +8,34 @@ , runConsumer ) where -import Relude+import Prelude import Blammo.Logging-import Control.Exception.Annotated.UnliftIO (AnnotatedException)+import Control.Arrow ((&&&))+import Control.Exception.Annotated.UnliftIO+ ( AnnotatedException+ , Exception+ , displayException+ ) import Control.Exception.Annotated.UnliftIO qualified as Annotated import Control.Lens (Lens', view)+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.List.NonEmpty qualified as NE+import Data.Map.Strict (Map) import Data.Map.Strict qualified as Map+import Data.Text (Text) import Data.Text qualified as T+import Data.Text.Encoding qualified as T import Freckle.App.Env (Timeout (..)) import Freckle.App.Env qualified as Env import Freckle.App.Kafka.Producer (envKafkaBrokerAddresses)+import GHC.Stack (HasCallStack, prettyCallStack) import Kafka.Consumer hiding ( Timeout , closeConsumer@@ -105,8 +120,8 @@ (fmap Map.fromList . Env.keyValues) "KAFKA_EXTRA_SUBSCRIPTION_PROPS" (Env.def mempty)- pure- $ KafkaConsumerConfig+ pure $+ KafkaConsumerConfig brokerAddresses consumerGroupId kafkaTopic@@ -158,8 +173,8 @@ displayException e = mconcat [ "Unable to decode JSON"- , "\n input: " <> decodeUtf8 (input e)- , "\n errors: " <> (errors e)+ , "\n input: " <> T.unpack (T.decodeUtf8 (input e))+ , "\n errors: " <> errors e ] runConsumer@@ -187,22 +202,22 @@ for_ (crValue =<< mRecord) $ \bs -> do a <-- inSpan "kafka.consumer.message.decode" defaultSpanArguments- $ either (Annotated.throw . KafkaMessageDecodeError bs) pure- $ eitherDecodeStrict bs+ inSpan "kafka.consumer.message.decode" defaultSpanArguments $+ either (Annotated.throw . KafkaMessageDecodeError bs) pure $+ eitherDecodeStrict bs inSpan "kafka.consumer.message.handle" defaultSpanArguments $ onMessage a where kTimeout = Kafka.Timeout $ timeoutMs pollTimeout handlers =- [ Annotated.Handler- $ logErrorNS "kafka"- . annotatedExceptionMessageFrom @KafkaError- (const "Error polling for message from Kafka")- , Annotated.Handler- $ logErrorNS "kafka"- . annotatedExceptionMessageFrom @KafkaMessageDecodeError- (const "Could not decode message value")+ [ Annotated.Handler $+ logErrorNS "kafka"+ . annotatedExceptionMessageFrom @KafkaError+ (const "Error polling for message from Kafka")+ , Annotated.Handler $+ logErrorNS "kafka"+ . annotatedExceptionMessageFrom @KafkaMessageDecodeError+ (const "Could not decode message value") ] -- | Like 'annotatedExceptionMessage', but use the supplied function to@@ -231,4 +246,4 @@ err -> Annotated.throw err ) $ pure- . Just+ . Just
library/Freckle/App/Kafka/Producer.hs view
@@ -11,7 +11,7 @@ , produceKeyedOn ) where -import Relude+import Prelude import Blammo.Logging ( Message ((:#))@@ -22,15 +22,19 @@ ) import Control.Exception.Annotated.UnliftIO qualified as Annotated import Control.Lens (Lens', lens, view)+import Control.Monad.IO.Class (liftIO)+import Control.Monad.Reader (MonadReader) import Data.Aeson (ToJSON, encode)+import Data.ByteString.Lazy qualified as BSL+import Data.Foldable (for_, toList) import Data.HashMap.Strict qualified as HashMap+import Data.List.NonEmpty (NonEmpty) import Data.List.NonEmpty qualified as NE import Data.Pool (Pool) import Data.Pool qualified as Pool import Data.Text qualified as T import Data.Time (NominalDiffTime) import Freckle.App.Env qualified as Env-import GHC.IO.Exception (userError) import Kafka.Producer import OpenTelemetry.Trace (SpanKind (..), defaultSpanArguments) import OpenTelemetry.Trace qualified as Trace@@ -104,13 +108,13 @@ -> KafkaProducerPoolConfig -> IO (Pool KafkaProducer) createKafkaProducerPool addresses config =- Pool.newPool- $ Pool.setNumStripes (Just $ kafkaProducerPoolConfigStripes config)- $ Pool.defaultPoolConfig- mkProducer- closeProducer- (realToFrac $ kafkaProducerPoolConfigIdleTimeout config)- (kafkaProducerPoolConfigSize config)+ Pool.newPool $+ Pool.setNumStripes (Just $ kafkaProducerPoolConfigStripes config) $+ Pool.defaultPoolConfig+ mkProducer+ closeProducer+ (realToFrac $ kafkaProducerPoolConfigIdleTimeout config)+ (kafkaProducerPoolConfigSize config) where mkProducer = either@@ -142,15 +146,17 @@ for_ @NonEmpty values $ \value -> do mError <- liftIO $ produceMessage producer $ mkProducerRecord value for_ @Maybe mError $ \e ->- run $ logErrorNS "kafka" $ "Failed to send event"- :# ["error" .= (show e :: Text)]+ run $+ logErrorNS "kafka" $+ "Failed to send event"+ :# ["error" .= T.pack (show e)] where mkProducerRecord value = ProducerRecord { prTopic , prPartition = UnassignedPartition- , prKey = Just $ toStrict $ encode $ keyF value- , prValue = Just $ toStrict $ encode value+ , prKey = Just $ BSL.toStrict $ encode $ keyF value+ , prValue = Just $ BSL.toStrict $ encode value , prHeaders = mempty }
package.yaml view
@@ -1,5 +1,5 @@ name: freckle-kafka-version: 0.0.0.0+version: 0.0.0.1 maintainer: Freckle Education category: Database github: freckle/freckle-app@@ -57,12 +57,13 @@ - Blammo >= 2.0.0.0 - aeson - annotated-exception+ - bytestring - containers - freckle-env - hs-opentelemetry-sdk - hw-kafka-client - lens- - relude+ - mtl - resource-pool >= 0.4.0.0 - text - time