diff --git a/hw-kafka-client.cabal b/hw-kafka-client.cabal --- a/hw-kafka-client.cabal +++ b/hw-kafka-client.cabal @@ -1,7 +1,7 @@ cabal-version: 2.2 name: hw-kafka-client -version: 3.1.2 +version: 4.0.0 synopsis: Kafka bindings for Haskell description: Apache Kafka bindings backed by the librdkafka C library. . @@ -19,7 +19,7 @@ maintainer: Alexey Raga license: MIT license-file: LICENSE -tested-with: GHC == 8.10.1, GHC == 8.8.3, GHC == 8.6.5, GHC == 8.4.4 +tested-with: GHC == 8.10.2, GHC == 8.8.3, GHC == 8.6.5, GHC == 8.4.4 build-type: Simple extra-source-files: README.md diff --git a/src/Kafka/Callbacks.hs b/src/Kafka/Callbacks.hs --- a/src/Kafka/Callbacks.hs +++ b/src/Kafka/Callbacks.hs @@ -2,11 +2,13 @@ ( errorCallback , logCallback , statsCallback +, Callback ) where +import Data.ByteString (ByteString) import Kafka.Internal.RdKafka (rdKafkaConfSetErrorCb, rdKafkaConfSetLogCb, rdKafkaConfSetStatsCb) -import Kafka.Internal.Setup (HasKafkaConf(..), getRdKafkaConf) +import Kafka.Internal.Setup (HasKafkaConf(..), getRdKafkaConf, Callback(..)) import Kafka.Types (KafkaError(..), KafkaLogLevel(..)) -- | Add a callback for errors. @@ -19,10 +21,10 @@ -- > -- > myErrorCallback :: 'KafkaError' -> String -> IO () -- > myErrorCallback kafkaError message = print $ show kafkaError <> "|" <> message -errorCallback :: HasKafkaConf k => (KafkaError -> String -> IO ()) -> k -> IO () -errorCallback callback k = +errorCallback :: (KafkaError -> String -> IO ()) -> Callback +errorCallback callback = let realCb _ err = callback (KafkaResponseError err) - in rdKafkaConfSetErrorCb (getRdKafkaConf k) realCb + in Callback $ \k -> rdKafkaConfSetErrorCb (getRdKafkaConf k) realCb -- | Add a callback for logs. -- @@ -34,12 +36,12 @@ -- > -- > myLogCallback :: 'KafkaLogLevel' -> String -> String -> IO () -- > myLogCallback level facility message = print $ show level <> "|" <> facility <> "|" <> message -logCallback :: HasKafkaConf k => (KafkaLogLevel -> String -> String -> IO ()) -> k -> IO () -logCallback callback k = +logCallback :: (KafkaLogLevel -> String -> String -> IO ()) -> Callback +logCallback callback = let realCb _ = callback . toEnum - in rdKafkaConfSetLogCb (getRdKafkaConf k) realCb + in Callback $ \k -> rdKafkaConfSetLogCb (getRdKafkaConf k) realCb --- | Add a callback for stats. +-- | Add a callback for stats. The passed ByteString contains an UTF-8 encoded JSON document and can e.g. be parsed using Data.Aeson.decodeStrict. For more information about the content of the JSON document see . -- -- ==== __Examples__ -- @@ -49,7 +51,7 @@ -- > -- > myStatsCallback :: String -> IO () -- > myStatsCallback stats = print $ show stats -statsCallback :: HasKafkaConf k => (String -> IO ()) -> k -> IO () -statsCallback callback k = +statsCallback :: (ByteString -> IO ()) -> Callback +statsCallback callback = let realCb _ = callback - in rdKafkaConfSetStatsCb (getRdKafkaConf k) realCb + in Callback $ \k -> rdKafkaConfSetStatsCb (getRdKafkaConf k) realCb diff --git a/src/Kafka/Consumer.hs b/src/Kafka/Consumer.hs --- a/src/Kafka/Consumer.hs +++ b/src/Kafka/Consumer.hs @@ -85,7 +85,7 @@ import Kafka.Consumer.Convert (fromMessagePtr, fromNativeTopicPartitionList'', offsetCommitToBool, offsetToInt64, toMap, toNativeTopicPartitionList, toNativeTopicPartitionList', toNativeTopicPartitionListNoDispose, topicPartitionFromMessageForCommit) import Kafka.Consumer.Types (KafkaConsumer (..)) import Kafka.Internal.RdKafka (RdKafkaRespErrT (..), RdKafkaTopicPartitionListTPtr, RdKafkaTypeT (..), newRdKafkaT, newRdKafkaTopicPartitionListT, newRdKafkaTopicT, rdKafkaAssign, rdKafkaAssignment, rdKafkaCommit, rdKafkaCommitted, rdKafkaConfSetDefaultTopicConf, rdKafkaConsumeBatchQueue, rdKafkaConsumeQueue, rdKafkaConsumerClose, rdKafkaConsumerPoll, rdKafkaOffsetsStore, rdKafkaPausePartitions, rdKafkaPollSetConsumer, rdKafkaPosition, rdKafkaQueueDestroy, rdKafkaQueueNew, rdKafkaResumePartitions, rdKafkaSeek, rdKafkaSetLogLevel, rdKafkaSubscribe, rdKafkaSubscription, rdKafkaTopicConfDup, rdKafkaTopicPartitionListAdd) -import Kafka.Internal.Setup (CallbackPollStatus (..), Kafka (..), KafkaConf (..), KafkaProps (..), TopicConf (..), TopicProps (..), getKafkaConf, getRdKafka, kafkaConf, topicConf) +import Kafka.Internal.Setup (CallbackPollStatus (..), Kafka (..), KafkaConf (..), KafkaProps (..), TopicConf (..), TopicProps (..), getKafkaConf, getRdKafka, kafkaConf, topicConf, Callback(..)) import Kafka.Internal.Shared (kafkaErrorToMaybe, maybeToLeft, rdKafkaErrorToEither) import Kafka.Consumer.ConsumerProperties as X @@ -327,7 +327,7 @@ newConsumerConf :: ConsumerProperties -> IO KafkaConf newConsumerConf ConsumerProperties {cpProps = m, cpCallbacks = cbs} = do conf <- kafkaConf (KafkaProps m) - forM_ cbs (\setCb -> setCb conf) + forM_ cbs (\(Callback setCb) -> setCb conf) return conf -- | Subscribes to a given list of topics. diff --git a/src/Kafka/Consumer/Callbacks.hs b/src/Kafka/Consumer/Callbacks.hs --- a/src/Kafka/Consumer/Callbacks.hs +++ b/src/Kafka/Consumer/Callbacks.hs @@ -14,16 +14,17 @@ import Kafka.Consumer.Convert (fromNativeTopicPartitionList', fromNativeTopicPartitionList'') import Kafka.Consumer.Types (KafkaConsumer (..), RebalanceEvent (..), TopicPartition (..)) import Kafka.Internal.RdKafka -import Kafka.Internal.Setup (HasKafka (..), HasKafkaConf (..), Kafka (..), KafkaConf (..), getRdMsgQueue) +import Kafka.Internal.Setup (HasKafka (..), HasKafkaConf (..), Kafka (..), KafkaConf (..), getRdMsgQueue, Callback (..)) import Kafka.Types (KafkaError (..), PartitionId (..), TopicName (..)) import qualified Data.Text as Text -- | Sets a callback that is called when rebalance is needed. -rebalanceCallback :: (KafkaConsumer -> RebalanceEvent -> IO ()) -> KafkaConf -> IO () -rebalanceCallback callback kc@(KafkaConf conf _ _) = rdKafkaConfSetRebalanceCb conf realCb +rebalanceCallback :: (KafkaConsumer -> RebalanceEvent -> IO ()) -> Callback +rebalanceCallback callback = + Callback $ \kc@(KafkaConf con _ _) -> rdKafkaConfSetRebalanceCb con (realCb kc) where - realCb k err pl = do + realCb kc k err pl = do k' <- newForeignPtr_ k pls <- newForeignPtr_ pl setRebalanceCallback callback (KafkaConsumer (Kafka k') kc) (KafkaResponseError err) pls @@ -36,10 +37,11 @@ -- If no partitions had valid offsets to commit this callback will be called -- with 'KafkaResponseError' 'RdKafkaRespErrNoOffset' which is not to be considered -- an error. -offsetCommitCallback :: (KafkaConsumer -> KafkaError -> [TopicPartition] -> IO ()) -> KafkaConf -> IO () -offsetCommitCallback callback kc@(KafkaConf conf _ _) = rdKafkaConfSetOffsetCommitCb conf realCb +offsetCommitCallback :: (KafkaConsumer -> KafkaError -> [TopicPartition] -> IO ()) -> Callback +offsetCommitCallback callback = + Callback $ \kc@(KafkaConf conf _ _) -> rdKafkaConfSetOffsetCommitCb conf (realCb kc) where - realCb k err pl = do + realCb kc k err pl = do k' <- newForeignPtr_ k pls <- fromNativeTopicPartitionList' pl callback (KafkaConsumer (Kafka k') kc) (KafkaResponseError err) pls diff --git a/src/Kafka/Consumer/ConsumerProperties.hs b/src/Kafka/Consumer/ConsumerProperties.hs --- a/src/Kafka/Consumer/ConsumerProperties.hs +++ b/src/Kafka/Consumer/ConsumerProperties.hs @@ -17,6 +17,7 @@ , logLevel , compression , suppressDisconnectLogs +, statisticsInterval , extraProps , extraProp , debugOptions @@ -33,7 +34,7 @@ import Data.Text (Text) import qualified Data.Text as Text import Kafka.Consumer.Types (ConsumerGroupId (..)) -import Kafka.Internal.Setup (KafkaConf (..)) +import Kafka.Internal.Setup (KafkaConf (..), Callback(..)) import Kafka.Types (BrokerAddress (..), ClientId (..), KafkaCompressionCodec (..), KafkaDebug (..), KafkaLogLevel (..), Millis (..), kafkaCompressionCodecToText, kafkaDebugToText) import Kafka.Consumer.Callbacks as X @@ -52,7 +53,7 @@ data ConsumerProperties = ConsumerProperties { cpProps :: Map Text Text , cpLogLevel :: Maybe KafkaLogLevel - , cpCallbacks :: [KafkaConf -> IO ()] + , cpCallbacks :: [Callback] , cpCallbackPollMode :: CallbackPollMode } @@ -116,7 +117,7 @@ -- * 'errorCallback' -- * 'logCallback' -- * 'statsCallback' -setCallback :: (KafkaConf -> IO ()) -> ConsumerProperties +setCallback :: Callback -> ConsumerProperties setCallback cb = mempty { cpCallbacks = [cb] } -- | Set the logging level. @@ -136,6 +137,11 @@ suppressDisconnectLogs :: ConsumerProperties suppressDisconnectLogs = extraProps $ M.fromList [("log.connection.close", "false")] + +-- | Set the for the producer. +statisticsInterval :: Millis -> ConsumerProperties +statisticsInterval (Millis t) = + extraProps $ M.singleton "statistics.interval.ms" (Text.pack $ show t) -- | Set any configuration options that are supported by /librdkafka/. -- The full list can be found diff --git a/src/Kafka/Internal/RdKafka.chs b/src/Kafka/Internal/RdKafka.chs --- a/src/Kafka/Internal/RdKafka.chs +++ b/src/Kafka/Internal/RdKafka.chs @@ -3,6 +3,8 @@ module Kafka.Internal.RdKafka where +import Data.ByteString (ByteString) +import qualified Data.ByteString as BS import Data.Text (Text) import qualified Data.Text as Text import Control.Monad (liftM) @@ -15,7 +17,7 @@ import Foreign.Ptr (Ptr, FunPtr, castPtr, nullPtr) import Foreign.ForeignPtr (FinalizerPtr, addForeignPtrFinalizer, newForeignPtr_, withForeignPtr) import Foreign.C.Error (Errno(..), getErrno) -import Foreign.C.String (CString, newCString, withCAString, peekCAString, peekCAStringLen, peekCString) +import Foreign.C.String (CString, newCString, withCAString, peekCAString, peekCString) import Foreign.C.Types (CFile, CInt(..), CSize, CChar) import System.IO (Handle, stdin, stdout, stderr) import System.Posix.IO (handleToFd) @@ -414,7 +416,7 @@ ---- Stats Callback type StatsCallback' = Ptr RdKafkaT -> CString -> CSize -> Word8Ptr -> IO () -type StatsCallback = Ptr RdKafkaT -> String -> IO () +type StatsCallback = Ptr RdKafkaT -> ByteString -> IO () foreign import ccall safe "wrapper" mkStatsCallback :: StatsCallback' -> IO (FunPtr StatsCallback') @@ -424,7 +426,7 @@ rdKafkaConfSetStatsCb :: RdKafkaConfTPtr -> StatsCallback -> IO () rdKafkaConfSetStatsCb conf cb = do - cb' <- mkStatsCallback $ \k j jl _ -> peekCAStringLen (j, cIntConv jl) >>= cb k + cb' <- mkStatsCallback $ \k j jl _ -> BS.packCStringLen (j, cIntConv jl) >>= cb k withForeignPtr conf $ \c -> rdKafkaConfSetStatsCb' c cb' return () diff --git a/src/Kafka/Internal/Setup.hs b/src/Kafka/Internal/Setup.hs --- a/src/Kafka/Internal/Setup.hs +++ b/src/Kafka/Internal/Setup.hs @@ -7,6 +7,7 @@ , HasKafka(..) , HasKafkaConf(..) , HasTopicConf(..) +, Callback(..) , CallbackPollStatus(..) , getRdKafka , getRdKafkaConf @@ -45,6 +46,11 @@ newtype TopicProps = TopicProps (Map Text Text) deriving (Show, Eq) newtype Kafka = Kafka RdKafkaTPtr deriving Show newtype TopicConf = TopicConf RdKafkaTopicConfTPtr deriving Show + +-- | Callbacks allow retrieving various information like error occurences, statistics +-- and log messages. +-- See `Kafka.Consumer.setCallback` (Consumer) and `Kafka.Producer.setCallback` (Producer) for more details. +newtype Callback = Callback (KafkaConf -> IO ()) data CallbackPollStatus = CallbackPollEnabled | CallbackPollDisabled deriving (Show, Eq) diff --git a/src/Kafka/Producer.hs b/src/Kafka/Producer.hs --- a/src/Kafka/Producer.hs +++ b/src/Kafka/Producer.hs @@ -82,7 +82,7 @@ import Foreign.Storable (Storable (..)) import Foreign.StablePtr (newStablePtr, castStablePtrToPtr) import Kafka.Internal.RdKafka (RdKafkaMessageT (..), RdKafkaRespErrT (..), RdKafkaTypeT (..), destroyUnmanagedRdKafkaTopic, newRdKafkaT, newUnmanagedRdKafkaTopicT, rdKafkaOutqLen, rdKafkaProduce, rdKafkaProduceBatch, rdKafkaSetLogLevel) -import Kafka.Internal.Setup (Kafka (..), KafkaConf (..), KafkaProps (..), TopicConf (..), TopicProps (..), kafkaConf, topicConf) +import Kafka.Internal.Setup (Kafka (..), KafkaConf (..), KafkaProps (..), TopicConf (..), TopicProps (..), kafkaConf, topicConf, Callback(..)) import Kafka.Internal.Shared (pollEvents) import Kafka.Producer.Convert (copyMsgFlags, handleProduceErr', producePartitionCInt, producePartitionInt) import Kafka.Producer.Types (KafkaProducer (..), ImmediateError(..)) @@ -120,7 +120,7 @@ deliveryCallback (const mempty) kc -- set callbacks - forM_ (ppCallbacks pps) (\setCb -> setCb kc) + forM_ (ppCallbacks pps) (\(Callback setCb) -> setCb kc) mbKafka <- newRdKafkaT RdKafkaProducer kc' case mbKafka of diff --git a/src/Kafka/Producer/ProducerProperties.hs b/src/Kafka/Producer/ProducerProperties.hs --- a/src/Kafka/Producer/ProducerProperties.hs +++ b/src/Kafka/Producer/ProducerProperties.hs @@ -12,6 +12,7 @@ , compression , topicCompression , sendTimeout +, statisticsInterval , extraProps , suppressDisconnectLogs , extraTopicProps @@ -26,8 +27,8 @@ import Data.Map (Map) import qualified Data.Map as M import Data.Semigroup as Sem -import Kafka.Internal.Setup (KafkaConf(..)) -import Kafka.Types (KafkaDebug(..), Timeout(..), KafkaCompressionCodec(..), KafkaLogLevel(..), BrokerAddress(..), kafkaDebugToText, kafkaCompressionCodecToText) +import Kafka.Internal.Setup (KafkaConf(..), Callback(..)) +import Kafka.Types (KafkaDebug(..), Timeout(..), KafkaCompressionCodec(..), KafkaLogLevel(..), BrokerAddress(..), kafkaDebugToText, kafkaCompressionCodecToText, Millis(..)) import Kafka.Producer.Callbacks @@ -36,7 +37,7 @@ { ppKafkaProps :: Map Text Text , ppTopicProps :: Map Text Text , ppLogLevel :: Maybe KafkaLogLevel - , ppCallbacks :: [KafkaConf -> IO ()] + , ppCallbacks :: [Callback] } instance Sem.Semigroup ProducerProperties where @@ -69,7 +70,7 @@ -- * 'errorCallback' -- * 'logCallback' -- * 'statsCallback' -setCallback :: (KafkaConf -> IO ()) -> ProducerProperties +setCallback :: Callback -> ProducerProperties setCallback cb = mempty { ppCallbacks = [cb] } -- | Sets the logging level. @@ -91,6 +92,11 @@ sendTimeout :: Timeout -> ProducerProperties sendTimeout (Timeout t) = extraTopicProps $ M.singleton "message.timeout.ms" (Text.pack $ show t) + +-- | Set the for the producer. +statisticsInterval :: Millis -> ProducerProperties +statisticsInterval (Millis t) = + extraProps $ M.singleton "statistics.interval.ms" (Text.pack $ show t) -- | Any configuration options that are supported by /librdkafka/. -- The full list can be found