diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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) diff --git a/freckle-kafka.cabal b/freckle-kafka.cabal --- a/freckle-kafka.cabal +++ b/freckle-kafka.cabal @@ -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, diff --git a/library/Freckle/App/Kafka/Consumer.hs b/library/Freckle/App/Kafka/Consumer.hs --- a/library/Freckle/App/Kafka/Consumer.hs +++ b/library/Freckle/App/Kafka/Consumer.hs @@ -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 diff --git a/library/Freckle/App/Kafka/Producer.hs b/library/Freckle/App/Kafka/Producer.hs --- a/library/Freckle/App/Kafka/Producer.hs +++ b/library/Freckle/App/Kafka/Producer.hs @@ -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 } diff --git a/package.yaml b/package.yaml --- a/package.yaml +++ b/package.yaml @@ -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