diff --git a/haskakafka.cabal b/haskakafka.cabal --- a/haskakafka.cabal +++ b/haskakafka.cabal @@ -1,5 +1,5 @@ name: haskakafka -version: 1.1.0 +version: 1.2.0 synopsis: Kafka bindings for Haskell description: Apache Kafka bindings backed by the librdkafka C library. This implies full consumer and producer @@ -26,8 +26,8 @@ , unix exposed-modules: Haskakafka - Haskakafka.InternalRdKafka Haskakafka.InternalRdKafkaEnum + Haskakafka.InternalRdKafka Haskakafka.InternalSetup Haskakafka.InternalTypes Haskakafka.ConsumerExample @@ -39,9 +39,9 @@ hs-source-dirs: src default-language: Haskell2010 ghc-options: -Wall - include-dirs: /usr/local/include/librdkafka - extra-lib-dirs: /usr/local/lib - extra-libraries: rdkafka + pkgconfig-depends: rdkafka + if os(darwin) + cpp-options: -D__attribute__(A)= -D_Nullable= -D_Nonnull= executable simple main-is: Simple.hs diff --git a/src/Haskakafka.hs b/src/Haskakafka.hs --- a/src/Haskakafka.hs +++ b/src/Haskakafka.hs @@ -8,8 +8,13 @@ , produceKeyedMessage , produceMessageBatch , storeOffset +, seekToOffset , getAllMetadata , getTopicMetadata +, handleProduceErr +, producePartitionInteger +, pollEvents +, pollEventsSafe -- Internal objects , IS.newKafka @@ -22,8 +27,8 @@ , rdKafkaVersionStr -- Type re-exports -, IT.Kafka -, IT.KafkaTopic +, IT.Kafka(..) +, IT.KafkaTopic(..) , IT.KafkaOffset(..) , IT.KafkaMessage(..) @@ -38,7 +43,7 @@ , IT.KafkaLogLevel(..) , IT.KafkaError(..) -, RDE.RdKafkaRespErrT +, RDE.RdKafkaRespErrT(..) -- Pseudo-internal , addBrokers @@ -82,13 +87,7 @@ -- 'consumeMessage' won't work without it. This function is non-blocking. startConsuming :: KafkaTopic -> Int -> KafkaOffset -> IO () startConsuming (KafkaTopic topicPtr _ _) partition offset = - let trueOffset = case offset of - KafkaOffsetBeginning -> (- 2) - KafkaOffsetEnd -> (- 1) - KafkaOffsetStored -> (- 1000) - KafkaOffsetInvalid -> (- 1001) - KafkaOffset i -> i - in throwOnError $ rdKafkaConsumeStart topicPtr partition trueOffset + throwOnError $ rdKafkaConsumeStart topicPtr partition $ offsetToInt64 offset -- | Stops consuming for a given topic. You probably do not need to call -- this directly (it is called automatically when 'withKafkaConsumer' terminates). @@ -134,6 +133,18 @@ [] -> return $ Right $ rights ms l -> return $ Left $ head l +seekToOffset :: KafkaTopic + -> Int -- ^ partition number + -> KafkaOffset -- ^ destination + -> Int -- ^ timeout in milliseconds + -> IO (Maybe KafkaError) +seekToOffset (KafkaTopic ptr _ _) p ofs timeout = do + err <- rdKafkaSeek ptr (fromIntegral p) + (fromIntegral $ offsetToInt64 ofs) timeout + case err of + RdKafkaRespErrNoError -> return Nothing + e -> return $ Just $ KafkaResponseError e + -- | Store a partition's offset in librdkafka's offset store. This function only needs to be called -- if auto.commit.enable is false. See -- for information on how to configure the offset store. @@ -216,6 +227,7 @@ , offset'RdKafkaMessageT = 0 , keyLen'RdKafkaMessageT = 0 , key'RdKafkaMessageT = nullPtr + , private'RdKafkaMessageT = nullPtr } produceMessageToMessage (KafkaProduceKeyedMessage kbs bs) = do let (payloadFPtr, payloadOffset, payloadLength) = BSI.toForeignPtr bs @@ -236,6 +248,7 @@ , offset'RdKafkaMessageT = 0 , keyLen'RdKafkaMessageT = keyLength , key'RdKafkaMessageT = passedKey + , private'RdKafkaMessageT = nullPtr } -- | Connects to Kafka broker in producer mode for a given topic, taking a function @@ -394,6 +407,12 @@ pollEvents :: Kafka -> Int -> IO () pollEvents (Kafka kPtr _) timeout = rdKafkaPoll kPtr timeout >> return () + +pollEventsSafe :: Kafka -> Int -> IO () +pollEventsSafe (Kafka kPtr _) timeout = do + _ <- withForeignPtr kPtr $ \realPtr -> do + rdKafkaPollSafe realPtr timeout + return () outboundQueueLength :: Kafka -> IO (Int) outboundQueueLength (Kafka kPtr _) = rdKafkaOutqLen kPtr diff --git a/src/Haskakafka/Consumer/Internal/Convert.hs b/src/Haskakafka/Consumer/Internal/Convert.hs --- a/src/Haskakafka/Consumer/Internal/Convert.hs +++ b/src/Haskakafka/Consumer/Internal/Convert.hs @@ -3,12 +3,11 @@ where import Control.Monad -import Data.Int import Foreign import Foreign.C.String import Haskakafka.Consumer.Internal.Types import Haskakafka.InternalRdKafka -import Haskakafka.InternalTypes +import Haskakafka.InternalShared -- | Converts offsets sync policy to integer (the way Kafka understands it): -- @@ -24,24 +23,6 @@ OffsetSyncImmediate -> 0 OffsetSyncInterval ms -> ms {-# INLINE offsetSyncToInt #-} - -offsetToInt64 :: KafkaOffset -> Int64 -offsetToInt64 o = case o of - KafkaOffsetBeginning -> -2 - KafkaOffsetEnd -> -1 - KafkaOffset off -> off - KafkaOffsetStored -> -1000 - KafkaOffsetInvalid -> -1001 -{-# INLINE offsetToInt64 #-} - -int64ToOffset :: Int64 -> KafkaOffset -int64ToOffset o - | o == -2 = KafkaOffsetBeginning - | o == -1 = KafkaOffsetEnd - | o == -1000 = KafkaOffsetStored - | o >= 0 = KafkaOffset o - | otherwise = KafkaOffsetInvalid -{-# INLINE int64ToOffset #-} fromNativeTopicPartitionList :: RdKafkaTopicPartitionListT -> IO [KafkaTopicPartition] fromNativeTopicPartitionList pl = diff --git a/src/Haskakafka/ConsumerExample.hs b/src/Haskakafka/ConsumerExample.hs --- a/src/Haskakafka/ConsumerExample.hs +++ b/src/Haskakafka/ConsumerExample.hs @@ -5,7 +5,7 @@ import Control.Arrow ((&&&)) import Haskakafka import Haskakafka.Consumer -import Haskakafka.InternalRdKafkaEnum +import Haskakafka.InternalRdKafkaEnum () iterator :: [Integer] iterator = [0 .. 20] diff --git a/src/Haskakafka/InternalRdKafka.chs b/src/Haskakafka/InternalRdKafka.chs --- a/src/Haskakafka/InternalRdKafka.chs +++ b/src/Haskakafka/InternalRdKafka.chs @@ -15,8 +15,10 @@ import System.Posix.IO import System.Posix.Types -#include "rdkafka.h" +#include "librdkafka/rdkafka.h" +{#import Haskakafka.InternalRdKafkaEnum#} + type CInt64T = {#type int64_t #} type CInt32T = {#type int32_t #} @@ -129,6 +131,7 @@ , offset'RdKafkaMessageT :: Int64 , payload'RdKafkaMessageT :: Word8Ptr , key'RdKafkaMessageT :: Word8Ptr + , private'RdKafkaMessageT :: Ptr () } deriving (Show, Eq) @@ -144,6 +147,7 @@ <*> liftM fromIntegral ({#get rd_kafka_message_t->offset #} p) <*> liftM castPtr ({#get rd_kafka_message_t->payload #} p) <*> liftM castPtr ({#get rd_kafka_message_t->key #} p) + <*> liftM castPtr ({#get rd_kafka_message_t->_private #} p) poke p x = do {#set rd_kafka_message_t.err#} p (enumToCInt $ err'RdKafkaMessageT x) {#set rd_kafka_message_t.rkt#} p (castPtr $ topic'RdKafkaMessageT x) @@ -153,6 +157,7 @@ {#set rd_kafka_message_t.offset#} p (fromIntegral $ offset'RdKafkaMessageT x) {#set rd_kafka_message_t.payload#} p (castPtr $ payload'RdKafkaMessageT x) {#set rd_kafka_message_t.key#} p (castPtr $ key'RdKafkaMessageT x) + {#set rd_kafka_message_t._private#} p (castPtr $ private'RdKafkaMessageT x) {#pointer *rd_kafka_message_t as RdKafkaMessageTPtr foreign -> RdKafkaMessageT #} @@ -167,7 +172,7 @@ instance Storable RdKafkaMetadataBrokerT where alignment _ = {#alignof rd_kafka_metadata_broker_t#} - sizeOf _ = 24 + sizeOf _ = {#sizeof rd_kafka_metadata_broker_t#} peek p = RdKafkaMetadataBrokerT <$> liftM fromIntegral ({#get rd_kafka_metadata_broker_t->id #} p) <*> liftM id ({#get rd_kafka_metadata_broker_t->host #} p) @@ -209,7 +214,7 @@ instance Storable RdKafkaMetadataTopicT where alignment _ = {#alignof rd_kafka_metadata_topic_t#} - sizeOf _ = 32 + sizeOf _ = {#sizeof rd_kafka_metadata_topic_t#} peek p = RdKafkaMetadataTopicT <$> liftM id ({#get rd_kafka_metadata_topic_t->topic #} p) <*> liftM fromIntegral ({#get rd_kafka_metadata_topic_t->partition_cnt #} p) @@ -514,9 +519,13 @@ {`RdKafkaTPtr', `RdKafkaMessageTPtr', `Int'} -> `RdKafkaRespErrT' cIntToEnum #} -{#fun unsafe rd_kafka_position as ^ +{#fun unsafe rd_kafka_committed as ^ {`RdKafkaTPtr', `RdKafkaTopicPartitionListTPtr', `Int'} -> `RdKafkaRespErrT' cIntToEnum #} + +{#fun unsafe rd_kafka_position as ^ + {`RdKafkaTPtr', `RdKafkaTopicPartitionListTPtr'} + -> `RdKafkaRespErrT' cIntToEnum #} ------------------------------------------------------------------------------------------------- ---- Groups data RdKafkaGroupMemberInfoT = RdKafkaGroupMemberInfoT @@ -623,6 +632,9 @@ foreign import ccall unsafe "rdkafka.h rd_kafka_message_destroy" rdKafkaMessageDestroy :: Ptr RdKafkaMessageT -> IO () +{#fun unsafe rd_kafka_message_timestamp as ^ + {`RdKafkaMessageTPtr', `RdKafkaTimestampTypeTPtr'} -> `CInt64T' cIntConv #} + -- rd_kafka_conf {#fun unsafe rd_kafka_conf_new as ^ {} -> `RdKafkaConfTPtr' #} @@ -720,6 +732,10 @@ {`RdKafkaTopicTPtr', cIntConv `CInt32T', `Int', castPtr `Ptr (Ptr RdKafkaMessageT)', cIntConv `CSize'} -> `CSize' cIntConv #} +{#fun rd_kafka_seek as ^ + {`RdKafkaTopicTPtr', cIntConv `CInt32T', cIntConv `CInt64T', `Int'} + -> `RdKafkaRespErrT' cIntToEnum #} + rdKafkaConsumeStop :: RdKafkaTopicTPtr -> Int -> IO (Maybe String) rdKafkaConsumeStop topicPtr partition = do i <- rdKafkaConsumeStopInternal topicPtr (fromIntegral partition) @@ -754,7 +770,10 @@ {# fun unsafe rd_kafka_metadata_destroy as ^ {castPtr `Ptr RdKafkaMetadataT'} -> `()' #} -{#fun unsafe rd_kafka_poll as ^ +foreign import ccall safe "rd_kafka.h rd_kafka_poll" + rdKafkaPollSafe :: Ptr RdKafkaT -> Int -> IO Int + +{# fun unsafe rd_kafka_poll as ^ {`RdKafkaTPtr', `Int'} -> `Int' #} {#fun unsafe rd_kafka_outq_len as ^ diff --git a/src/Haskakafka/InternalRdKafkaEnum.chs b/src/Haskakafka/InternalRdKafkaEnum.chs --- a/src/Haskakafka/InternalRdKafkaEnum.chs +++ b/src/Haskakafka/InternalRdKafkaEnum.chs @@ -3,8 +3,11 @@ module Haskakafka.InternalRdKafkaEnum where -#include "rdkafka.h" +#include "librdkafka/rdkafka.h" {#enum rd_kafka_type_t as ^ {underscoreToCase} deriving (Show, Eq) #} {#enum rd_kafka_conf_res_t as ^ {underscoreToCase} deriving (Show, Eq) #} {#enum rd_kafka_resp_err_t as ^ {underscoreToCase} deriving (Show, Eq) #} +{#enum rd_kafka_timestamp_type_t as ^ {underscoreToCase} deriving (Show, Eq) #} + +{#pointer *rd_kafka_timestamp_type_t as RdKafkaTimestampTypeTPtr foreign -> RdKafkaTimestampTypeT #} diff --git a/src/Haskakafka/InternalShared.hs b/src/Haskakafka/InternalShared.hs --- a/src/Haskakafka/InternalShared.hs +++ b/src/Haskakafka/InternalShared.hs @@ -66,3 +66,22 @@ _ -> Left err {-# INLINE kafkaErrorToEither #-} +offsetToInt64 :: KafkaOffset -> Int64 +offsetToInt64 o = case o of + KafkaOffsetBeginning -> -2 + KafkaOffsetEnd -> -1 + KafkaOffset off -> off + KafkaOffsetStored -> -1000 + KafkaOffsetInvalid -> -1001 + KafkaOffsetTail i -> -2000 - i +{-# INLINE offsetToInt64 #-} + +int64ToOffset :: Int64 -> KafkaOffset +int64ToOffset o + | o == -2 = KafkaOffsetBeginning + | o == -1 = KafkaOffsetEnd + | o == -1000 = KafkaOffsetStored + | o >= 0 = KafkaOffset o + | o <= -2000 = KafkaOffsetTail (-2000 - o) + | otherwise = KafkaOffsetInvalid +{-# INLINE int64ToOffset #-} diff --git a/src/Haskakafka/InternalTypes.hs b/src/Haskakafka/InternalTypes.hs --- a/src/Haskakafka/InternalTypes.hs +++ b/src/Haskakafka/InternalTypes.hs @@ -49,6 +49,7 @@ -- -- for offset store configuration. | KafkaOffsetStored + | KafkaOffsetTail Int64 | KafkaOffsetInvalid deriving (Eq, Show) diff --git a/tests/TestMain.hs b/tests/TestMain.hs --- a/tests/TestMain.hs +++ b/tests/TestMain.hs @@ -74,24 +74,25 @@ describe "Kafka topic configuration" $ do it "should allow dumping" $ do - kConf <- newKafkaTopicConf - kvs <- dumpKafkaTopicConf kConf + kTopicConf <- newKafkaTopicConf + kvs <- dumpKafkaTopicConf kTopicConf (Map.size kvs) `shouldSatisfy` (>0) it "should change when set is called" $ do - kConf <- newKafkaTopicConf - setKafkaTopicConfValue kConf "request.timeout.ms" "20000" - kvs <- dumpKafkaTopicConf kConf + kTopicConf <- newKafkaTopicConf + setKafkaTopicConfValue kTopicConf "request.timeout.ms" "20000" + kvs <- dumpKafkaTopicConf kTopicConf (kvs Map.! "request.timeout.ms") `shouldBe` "20000" it "should throw an exception on unknown property" $ do - kConf <- newKafkaTopicConf - (setKafkaTopicConfValue kConf "blippity.blop.cosby" "120") `shouldThrow` + kTopicConf <- newKafkaTopicConf + (setKafkaTopicConfValue kTopicConf "blippity.blop.cosby" "120") `shouldThrow` (\(KafkaUnknownConfigurationKey str) -> (length str) > 0) it "should throw an exception on an invalid value" $ do - kConf <- newKafkaTopicConf - (setKafkaTopicConfValue kConf "request.timeout.ms" "mono...doh!") `shouldThrow` + kTopicConf <- newKafkaTopicConf + (setKafkaTopicConfValue kTopicConf "request.timeout.ms" "mono...doh!") + `shouldThrow` (\(KafkaInvalidConfigurationValue str) -> (length str) > 0) describe "Logging" $ do @@ -150,7 +151,7 @@ produceMessageBatch producerTopic (KafkaSpecifiedPartition 0 ) sampleProduceMessages errs `shouldBe` [] - et <- consumeMessageBatch topic 0 (5000) 3 + et <- consumeMessageBatch topic 0 kafkaDelay 3 case et of (Left err) -> error $ show err (Right oms) -> do @@ -161,7 +162,7 @@ it "should not fail on batch consume when no messages are available #12" $ getAddressTopic $ \a t -> do withKafkaConsumer [] [] a t 0 KafkaOffsetEnd $ \_ topic -> do primeEOF topic - et <- consumeMessageBatch topic 0 (5000) 3 + et <- consumeMessageBatch topic 0 kafkaDelay 3 case et of (Left err) -> error $ show err (Right oms) -> do @@ -169,7 +170,7 @@ it "should return EOF on batch consume if necessary" $ getAddressTopic $ \a t -> do withKafkaConsumer [] [] a t 0 KafkaOffsetEnd $ \_ topic -> do - et <- consumeMessageBatch topic 0 (5000) 10 + et <- consumeMessageBatch topic 0 kafkaDelay 10 case et of (Left err) -> print err (Right _oms) -> error "should return EOF"