hw-kafka-client 2.1.3 → 2.2.0
raw patch · 6 files changed
+80/−25 lines, 6 files
Files
- example/ProducerExample.hs +25/−10
- hw-kafka-client.cabal +3/−1
- src/Kafka/Internal/RdKafka.chs +5/−4
- src/Kafka/Producer.hs +10/−5
- src/Kafka/Producer/Callbacks.hs +28/−0
- src/Kafka/Producer/ProducerProperties.hs +9/−5
example/ProducerExample.hs view
@@ -3,14 +3,17 @@ module ProducerExample where -import Control.Monad (forM_)-import Data.ByteString (ByteString)+import Control.Monad (forM_)+import Data.ByteString (ByteString)+import Data.ByteString.Char8 (pack) import Data.Monoid import Kafka.Producer -- Global producer properties producerProps :: ProducerProperties producerProps = brokersList [BrokerAddress "localhost:9092"]+ <> sendTimeout (Timeout 10000)+ <> setCallback (deliveryErrorsCallback print) <> logLevel KafkaLogDebug -- Topic to send messages to@@ -33,17 +36,29 @@ sendMessages :: KafkaProducer -> IO (Either KafkaError ()) sendMessages prod = do- err1 <- produceMessage prod (mkMessage Nothing (Just "test from producer") )+ putStrLn "Producer is ready, send your messages!"+ msg1 <- getLine++ err1 <- produceMessage prod (mkMessage Nothing (Just $ pack msg1)) forM_ err1 print - err2 <- produceMessage prod (mkMessage (Just "key") (Just "test from producer (with key)"))+ putStrLn "One more time!"+ msg2 <- getLine++ err2 <- produceMessage prod (mkMessage (Just "key") (Just $ pack msg2)) forM_ err2 print - errs <- produceMessageBatch prod- [ mkMessage (Just "b-1") (Just "batch-1")- , mkMessage (Just "b-2") (Just "batch-2")- , mkMessage Nothing (Just "batch-3")- ]+ putStrLn "And the last one..."+ msg3 <- getLine+ err3 <- produceMessage prod (mkMessage (Just "key3") (Just $ pack msg3)) - forM_ errs (print . snd)+ -- errs <- produceMessageBatch prod+ -- [ mkMessage (Just "b-1") (Just "batch-1")+ -- , mkMessage (Just "b-2") (Just "batch-2")+ -- , mkMessage Nothing (Just "batch-3")+ -- ]++ -- forM_ errs (print . snd)++ putStrLn "Thank you." return $ Right ()
hw-kafka-client.cabal view
@@ -1,5 +1,5 @@ name: hw-kafka-client-version: 2.1.3+version: 2.2.0 homepage: https://github.com/haskell-works/hw-kafka-client bug-reports: https://github.com/haskell-works/hw-kafka-client/issues license: MIT@@ -46,6 +46,7 @@ , transformers , unix , hw-kafka-client+ ghc-options: -threaded -rtsopts if flag(examples) buildable: True@@ -84,6 +85,7 @@ Kafka.Internal.Setup Kafka.Internal.Shared Kafka.Producer.Convert+ Kafka.Producer.Callbacks hs-source-dirs: src default-language: Haskell2010 ghc-options: -Wall
src/Kafka/Internal/RdKafka.chs view
@@ -296,17 +296,18 @@ return () ---- Delivery Callback-type DeliveryCallback = Ptr RdKafkaT -> Ptr RdKafkaMessageT -> Word8Ptr -> IO ()+type DeliveryCallback' = Ptr RdKafkaT -> Ptr RdKafkaMessageT -> Word8Ptr -> IO ()+type DeliveryCallback = Ptr RdKafkaT -> Ptr RdKafkaMessageT -> IO () foreign import ccall safe "wrapper"- mkDeliveryCallback :: DeliveryCallback -> IO (FunPtr DeliveryCallback)+ mkDeliveryCallback :: DeliveryCallback' -> IO (FunPtr DeliveryCallback') foreign import ccall safe "rd_kafka.h rd_kafka_conf_set_dr_msg_cb"- rdKafkaConfSetDrMsgCb' :: Ptr RdKafkaConfT -> FunPtr DeliveryCallback -> IO ()+ rdKafkaConfSetDrMsgCb' :: Ptr RdKafkaConfT -> FunPtr DeliveryCallback' -> IO () rdKafkaConfSetDrMsgCb :: RdKafkaConfTPtr -> DeliveryCallback -> IO () rdKafkaConfSetDrMsgCb conf cb = do- cb' <- mkDeliveryCallback cb+ cb' <- mkDeliveryCallback (\k m _ -> cb k m) withForeignPtr conf $ \c -> rdKafkaConfSetDrMsgCb' c cb' return ()
src/Kafka/Producer.hs view
@@ -26,6 +26,7 @@ import Kafka.Internal.RdKafka import Kafka.Internal.Setup import Kafka.Internal.Shared+import Kafka.Producer.Callbacks import Kafka.Producer.Convert import Kafka.Producer.Types @@ -55,7 +56,7 @@ -- A newly created producer must be closed with 'closeProducer' function. newProducer :: MonadIO m => ProducerProperties -> m (Either KafkaError KafkaProducer) newProducer pps = liftIO $ do- kc@(KafkaConf kc' ct) <- kafkaConf (KafkaProps $ M.toList (ppKafkaProps pps))+ kc@(KafkaConf kc' _) <- kafkaConf (KafkaProps $ M.toList (ppKafkaProps pps)) tc <- topicConf (TopicProps $ M.toList (ppTopicProps pps)) -- set callbacks@@ -67,7 +68,7 @@ Right kafka -> do forM_ (ppLogLevel pps) (rdKafkaSetLogLevel kafka . fromEnum) let prod = KafkaProducer (Kafka kafka) kc tc- runEventLoop prod ct (Just $ Timeout 100) >> return (Right prod)+ return (Right prod) -- | Sends a single message. -- Since librdkafka is backed by a queue, this function can return before messages are sent. See@@ -76,7 +77,8 @@ => KafkaProducer -> ProducerRecord -> m (Maybe KafkaError)-produceMessage (KafkaProducer (Kafka k) _ (TopicConf tc)) m = liftIO $+produceMessage kp@(KafkaProducer (Kafka k) _ (TopicConf tc)) m = liftIO $ do+ pollEvents kp (Just $ Timeout 0) -- fire callbacks if any exist (handle delivery reports) bracket (mkTopic $ prTopic m) clTopic withTopic where mkTopic (TopicName tn) = newUnmanagedRdKafkaTopicT k tn (Just tc)@@ -103,7 +105,8 @@ -> m [(ProducerRecord, KafkaError)] -- ^ An empty list when the operation is successful, -- otherwise a list of "failed" messages with corresponsing errors.-produceMessageBatch (KafkaProducer (Kafka k) _ (TopicConf tc)) messages = liftIO $+produceMessageBatch kp@(KafkaProducer (Kafka k) _ (TopicConf tc)) messages = liftIO $ do+ pollEvents kp (Just $ Timeout 0) -- fire callbacks if any exist (handle delivery reports) concat <$> forM (mkBatches messages) sendBatch where mkSortKey = prTopic &&& prPartition@@ -158,7 +161,9 @@ flushProducer kp = liftIO $ do pollEvents kp (Just $ Timeout 100) l <- outboundQueueLength (kpKafkaPtr kp)- unless (l == 0) $ flushProducer kp+ if (l == 0)+ then pollEvents kp (Just $ Timeout 0) -- to be sure that all the delivery reports are fired+ else flushProducer kp ------------------------------------------------------------------------------------
+ src/Kafka/Producer/Callbacks.hs view
@@ -0,0 +1,28 @@+module Kafka.Producer.Callbacks+( deliveryErrorsCallback+, module X+)+where++import Foreign+import Foreign.C.Error+import Kafka.Callbacks as X+import Kafka.Internal.RdKafka+import Kafka.Internal.Setup+import Kafka.Internal.Shared+import Kafka.Types++-- | Sets the callback for delivery errors.+-- The callback is only called in case of errors.+deliveryErrorsCallback :: (KafkaError -> IO ()) -> KafkaConf -> IO ()+deliveryErrorsCallback callback (KafkaConf conf _) = rdKafkaConfSetDrMsgCb conf realCb+ where+ realCb :: t -> Ptr RdKafkaMessageT -> IO ()+ realCb _ mptr =+ if mptr == nullPtr+ then getErrno >>= (callback . kafkaRespErr)+ else do+ s <- peek mptr+ if err'RdKafkaMessageT s /= RdKafkaRespErrNoError+ then (callback . KafkaResponseError $ err'RdKafkaMessageT s)+ else pure ()
src/Kafka/Producer/ProducerProperties.hs view
@@ -1,15 +1,15 @@ module Kafka.Producer.ProducerProperties ( module Kafka.Producer.ProducerProperties-, module Kafka.Callbacks+, module Kafka.Producer.Callbacks ) where import Control.Monad-import qualified Data.List as L-import Data.Map (Map)-import qualified Data.Map as M-import Kafka.Callbacks+import qualified Data.List as L+import Data.Map (Map)+import qualified Data.Map as M import Kafka.Internal.Setup+import Kafka.Producer.Callbacks import Kafka.Types -- | Properties to create 'KafkaProducer'.@@ -53,6 +53,10 @@ topicCompression :: KafkaCompressionCodec -> ProducerProperties topicCompression c = extraTopicProps $ M.singleton "compression.codec" (kafkaCompressionCodecToString c)++sendTimeout :: Timeout -> ProducerProperties+sendTimeout (Timeout t) =+ extraTopicProps $ M.singleton "message.timeout.ms" (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>