hw-kafka-client 3.1.2 → 4.0.0

raw patch · 9 files changed

+58/−34 lines, 9 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

• hw-kafka-client.cabal +2/−2
• src/Kafka/Callbacks.hs +13/−11
• src/Kafka/Consumer.hs +2/−2
• src/Kafka/Consumer/Callbacks.hs +9/−7
• src/Kafka/Consumer/ConsumerProperties.hs +9/−3
• src/Kafka/Internal/RdKafka.chs +5/−3
• src/Kafka/Internal/Setup.hs +6/−0
• src/Kafka/Producer.hs +2/−2
• src/Kafka/Producer/ProducerProperties.hs +10/−4

@@ -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 <alexey.raga@gmail.com> 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 

@@ -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 <https://github.com/edenhill/librdkafka/blob/master/STATISTICS.md>. -- -- ==== __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

@@ -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.

@@ -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

@@ -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 <https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md statistics.interval.ms> 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 <https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md here>

@@ -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 () 

@@ -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) 

@@ -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

@@ -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 <https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md statistics.interval.ms> 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 <https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md here>