-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
* Kafka consumer module Transplanted from: freckle/progress#42 * Documentation * Inline `keyValueMap` This isn't a novel enough function to warrant inclusion into `Freckle.App.Env` * Abstract `timeout` from Database This is planned to be used for reading in the polling timeout setting in progress consumer * Formatting * Fix doctest * Use `Timeout` in `runConsumer` * Use `logDebug` * Move Kafka to Producer module * `v1.9.2.0` * Poll single topic * Commit offset only after processing valid message
- Loading branch information
Showing
8 changed files
with
377 additions
and
190 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,153 +1,7 @@ | ||
| {-# LANGUAGE ApplicativeDo #-} | ||
| {-# LANGUAGE NamedFieldPuns #-} | ||
|
|
||
| module Freckle.App.Kafka | ||
| ( envKafkaBrokerAddresses | ||
| , KafkaProducerPoolConfig (..) | ||
| , envKafkaProducerPoolConfig | ||
| , KafkaProducerPool (..) | ||
| , HasKafkaProducerPool (..) | ||
| , createKafkaProducerPool | ||
| , produceKeyedOn | ||
| , produceKeyedOnAsync | ||
| ( module Freckle.App.Kafka.Consumer | ||
| , module Freckle.App.Kafka.Producer | ||
| ) where | ||
|
|
||
| import Freckle.App.Prelude | ||
|
|
||
| import Blammo.Logging | ||
| import Control.Lens (Lens', view) | ||
| import Data.Aeson (ToJSON, encode) | ||
| import Data.ByteString.Lazy (toStrict) | ||
| import qualified Data.List.NonEmpty as NE | ||
| import Data.Pool (Pool) | ||
| import qualified Data.Pool as Pool | ||
| import qualified Data.Text as T | ||
| import qualified Freckle.App.Env as Env | ||
| import Kafka.Producer | ||
| import UnliftIO.Async (async) | ||
| import UnliftIO.Exception (throwString) | ||
| import Yesod.Core.Lens | ||
| import Yesod.Core.Types (HandlerData) | ||
|
|
||
| envKafkaBrokerAddresses | ||
| :: Env.Parser Env.Error (NonEmpty BrokerAddress) | ||
| envKafkaBrokerAddresses = | ||
| Env.var | ||
| (Env.eitherReader readKafkaBrokerAddresses) | ||
| "KAFKA_BROKER_ADDRESSES" | ||
| mempty | ||
|
|
||
| readKafkaBrokerAddresses :: String -> Either String (NonEmpty BrokerAddress) | ||
| readKafkaBrokerAddresses t = case NE.nonEmpty $ T.splitOn "," $ T.pack t of | ||
| Just xs@(x NE.:| _) | ||
| | x /= "" -> Right $ BrokerAddress <$> xs | ||
| _ -> Left "Broker Address cannot be empty" | ||
|
|
||
| data KafkaProducerPoolConfig = KafkaProducerPoolConfig | ||
| { kafkaProducerPoolConfigStripes :: Int | ||
| -- ^ The number of stripes (distinct sub-pools) to maintain. | ||
| -- The smallest acceptable value is 1. | ||
| , kafkaProducerPoolConfigIdleTimeout :: NominalDiffTime | ||
| -- ^ Amount of time for which an unused resource is kept open. | ||
| -- The smallest acceptable value is 0.5 seconds. | ||
| -- | ||
| -- The elapsed time before destroying a resource may be a little | ||
| -- longer than requested, as the reaper thread wakes at 1-second | ||
| -- intervals. | ||
| , kafkaProducerPoolConfigSize :: Int | ||
| -- ^ Maximum number of resources to keep open per stripe. The | ||
| -- smallest acceptable value is 1. | ||
| -- | ||
| -- Requests for resources will block if this limit is reached on a | ||
| -- single stripe, even if other stripes have idle resources | ||
| -- available. | ||
| } | ||
| deriving stock (Show) | ||
|
|
||
| -- | Same defaults as 'Database.Persist.Sql.ConnectionPoolConfig' | ||
| defaultKafkaProducerPoolConfig :: KafkaProducerPoolConfig | ||
| defaultKafkaProducerPoolConfig = KafkaProducerPoolConfig 1 600 10 | ||
|
|
||
| envKafkaProducerPoolConfig | ||
| :: Env.Parser Env.Error KafkaProducerPoolConfig | ||
| envKafkaProducerPoolConfig = do | ||
| poolSize <- Env.var Env.auto "KAFKA_PRODUCER_POOL_SIZE" $ Env.def 10 | ||
| pure $ defaultKafkaProducerPoolConfig {kafkaProducerPoolConfigSize = poolSize} | ||
|
|
||
| data KafkaProducerPool | ||
| = NullKafkaProducerPool | ||
| | KafkaProducerPool (Pool KafkaProducer) | ||
|
|
||
| class HasKafkaProducerPool env where | ||
| kafkaProducerPoolL :: Lens' env KafkaProducerPool | ||
|
|
||
| instance HasKafkaProducerPool site => HasKafkaProducerPool (HandlerData child site) where | ||
| kafkaProducerPoolL = envL . siteL . kafkaProducerPoolL | ||
|
|
||
| createKafkaProducerPool | ||
| :: NonEmpty BrokerAddress | ||
| -> KafkaProducerPoolConfig | ||
| -> IO (Pool KafkaProducer) | ||
| createKafkaProducerPool addresses KafkaProducerPoolConfig {..} = | ||
| Pool.createPool | ||
| mkProducer | ||
| closeProducer | ||
| kafkaProducerPoolConfigStripes | ||
| kafkaProducerPoolConfigIdleTimeout | ||
| kafkaProducerPoolConfigSize | ||
| where | ||
| mkProducer = | ||
| either throw pure =<< newProducer (brokersList $ toList addresses) | ||
| throw err = throwString $ "Failed to open kafka producer: " <> show err | ||
|
|
||
| produceKeyedOn | ||
| :: ( ToJSON value | ||
| , ToJSON key | ||
| , MonadLogger m | ||
| , MonadReader env m | ||
| , HasKafkaProducerPool env | ||
| , MonadUnliftIO m | ||
| ) | ||
| => TopicName | ||
| -> NonEmpty value | ||
| -> (value -> key) | ||
| -> m () | ||
| produceKeyedOn prTopic values keyF = do | ||
| logDebugNS "kafka" $ "Producing Kafka events" :# ["events" .= values] | ||
| view kafkaProducerPoolL >>= \case | ||
| NullKafkaProducerPool -> pure () | ||
| KafkaProducerPool producerPool -> do | ||
| errors <- | ||
| liftIO $ | ||
| Pool.withResource producerPool $ \producer -> | ||
| produceMessageBatch producer $ | ||
| toList $ | ||
| mkProducerRecord <$> values | ||
| unless (null errors) $ | ||
| logErrorNS "kafka" $ | ||
| "Failed to send events" :# ["errors" .= fmap (tshow . snd) errors] | ||
| where | ||
| mkProducerRecord value = | ||
| ProducerRecord | ||
| { prTopic | ||
| , prPartition = UnassignedPartition | ||
| , prKey = Just $ toStrict $ encode $ keyF value | ||
| , prValue = | ||
| Just $ | ||
| toStrict $ | ||
| encode value | ||
| } | ||
|
|
||
| produceKeyedOnAsync | ||
| :: ( ToJSON value | ||
| , ToJSON key | ||
| , MonadLogger m | ||
| , MonadReader env m | ||
| , HasKafkaProducerPool env | ||
| , MonadUnliftIO m | ||
| ) | ||
| => TopicName | ||
| -> NonEmpty value | ||
| -> (value -> key) | ||
| -> m () | ||
| produceKeyedOnAsync prTopic values = void . async . produceKeyedOn prTopic values | ||
| import Freckle.App.Kafka.Consumer | ||
| import Freckle.App.Kafka.Producer |
Oops, something went wrong.