haskakafka 0.2.0.2 → 0.5.0
raw patch · 5 files changed
+482/−110 lines, 5 filesdep +either-unwrapdep +regex-posixdep +temporarydep ~base
Dependencies added: either-unwrap, regex-posix, temporary
Dependency ranges changed: base
Files
- haskakafka.cabal +12/−8
- src/Haskakafka.hs +205/−45
- src/Haskakafka/Example.hs +64/−0
- src/Haskakafka/Internal.chs +33/−10
- tests/TestMain.hs +168/−47
haskakafka.cabal view
@@ -1,9 +1,8 @@ name: haskakafka-version: 0.2.0.2+version: 0.5.0 synopsis: Kafka bindings for Haskell description: Use Apache Kafka in Haskell through the librdkafka- C library. The library is preliminary but fully- functional. + C library. homepage: http://github.com/cosbynator/haskakafka license: MIT license-file: LICENSE@@ -21,14 +20,16 @@ library Build-tools: c2hs- build-depends: base >=4.7 && < 5- , containers- , unix+ build-depends: base >=4.6 && < 5 , bytestring+ , containers + , temporary+ , unix exposed-modules: Haskakafka Haskakafka.Internal Haskakafka.InternalEnum+ Haskakafka.Example hs-source-dirs: src default-language: Haskell2010 ghc-options: -Wall @@ -42,7 +43,10 @@ Main-Is: TestMain.hs HS-Source-Dirs: tests ghc-options: -Wall -threaded- build-depends: base >=4.7 && < 5- , hspec+ build-depends: base >=4.6 && < 5 , bytestring+ , containers , haskakafka+ , hspec+ , regex-posix+ , either-unwrap
src/Haskakafka.hs view
@@ -3,6 +3,7 @@ module Haskakafka ( Kafka , KafkaConf+ , KafkaLogLevel(..) , KafkaMessage(..) , KafkaOffset(..) , KafkaMetadata(..)@@ -15,9 +16,14 @@ , KafkaTopicConf , KafkaType(..) , KafkaError (..)+ , rdKafkaVersionStr+ , supportedKafkaConfProperties , hPrintKafkaProperties , hPrintKafka , newKafkaConf+ , setKafkaConfValue+ , setKafkaTopicConfValue+ , setKafkaLogLevel , dumpKafkaConf , newKafkaTopicConf , newKafka@@ -26,13 +32,20 @@ , addBrokers , startConsuming , consumeMessage+ , consumeMessageBatch+ , storeOffset , stopConsuming , produceMessage+ , produceKeyedMessage , produceMessageBatch , pollEvents , drainOutQueue , getAllMetadata , getTopicMetadata+ , withKafkaProducer+ , withKafkaConsumer+ , dumpConfFromKafkaTopic+ , dumpConfFromKafka , module Haskakafka.InternalEnum ) where @@ -47,15 +60,43 @@ import Haskakafka.Internal import Haskakafka.InternalEnum import System.IO+import System.IO.Temp (withSystemTempFile) import qualified Data.Map.Strict as Map import qualified Data.ByteString as BS import qualified Data.ByteString.Internal as BSI +data KafkaLogLevel = + KafkaLogEmerg | KafkaLogAlert | KafkaLogCrit | KafkaLogErr | KafkaLogWarning |+ KafkaLogNotice | KafkaLogInfo | KafkaLogDebug++instance Enum KafkaLogLevel where+ toEnum 0 = KafkaLogEmerg+ toEnum 1 = KafkaLogAlert+ toEnum 2 = KafkaLogCrit+ toEnum 3 = KafkaLogErr+ toEnum 4 = KafkaLogWarning+ toEnum 5 = KafkaLogNotice+ toEnum 6 = KafkaLogInfo+ toEnum 7 = KafkaLogDebug+ toEnum _ = undefined++ fromEnum KafkaLogEmerg = 0+ fromEnum KafkaLogAlert = 1+ fromEnum KafkaLogCrit = 2+ fromEnum KafkaLogErr = 3+ fromEnum KafkaLogWarning = 4+ fromEnum KafkaLogNotice = 5+ fromEnum KafkaLogInfo = 6+ fromEnum KafkaLogDebug = 7+ data KafkaError = KafkaError String | KafkaInvalidReturnValue | KafkaBadSpecification String | KafkaResponseError RdKafkaRespErrT+ | KafkaInvalidConfigurationValue String+ | KafkaUnknownConfigurationKey String+ | KakfaBadConfiguration deriving (Eq, Show, Typeable) instance Exception KafkaError@@ -86,20 +127,28 @@ | KafkaOffsetStored | KafkaOffset Int64 +data KafkaConf = KafkaConf RdKafkaConfTPtr+data KafkaTopicConf = KafkaTopicConf RdKafkaTopicConfTPtr+ data KafkaType = KafkaConsumer | KafkaProducer-data Kafka = Kafka { kafkaPtr :: RdKafkaTPtr}+data Kafka = Kafka { kafkaPtr :: RdKafkaTPtr, _kafkaConf :: KafkaConf} data KafkaTopic = KafkaTopic - { kafkaTopicPtr :: RdKafkaTopicTPtr - , owningKafka :: Kafka -- prevents garbage collection - } + RdKafkaTopicTPtr + Kafka -- Kept around to prevent garbage collection + KafkaTopicConf -data KafkaConf = KafkaConf {kafkaConfPtr :: RdKafkaConfTPtr}-data KafkaTopicConf = KafkaTopicConf {kafkaTopicConfPtr :: RdKafkaTopicConfTPtr} kafkaTypeToRdKafkaType :: KafkaType -> RdKafkaTypeT kafkaTypeToRdKafkaType KafkaConsumer = RdKafkaConsumer kafkaTypeToRdKafkaType KafkaProducer = RdKafkaProducer +-- Because of the rdKafka API, we have to create a temp file to get properties into a string+supportedKafkaConfProperties :: IO (String)+supportedKafkaConfProperties = do + withSystemTempFile "haskakafka.rdkafka_properties" $ \tP tH -> do+ hPrintKafkaProperties tH+ readFile tP+ hPrintKafkaProperties :: Handle -> IO () hPrintKafkaProperties h = handleToCFile h "w" >>= rdKafkaConfPropertiesShow @@ -112,32 +161,58 @@ newKafkaConf :: IO KafkaConf newKafkaConf = newRdKafkaConfT >>= return . KafkaConf +checkConfSetValue :: RdKafkaConfResT -> CCharBufPointer -> IO ()+checkConfSetValue err charPtr = case err of + RdKafkaConfOk -> return ()+ RdKafkaConfInvalid -> do + str <- peekCString charPtr+ throw $ KafkaInvalidConfigurationValue str+ RdKafkaConfUnknown -> do+ str <- peekCString charPtr+ throw $ KafkaUnknownConfigurationKey str++setKafkaConfValue :: KafkaConf -> String -> String -> IO ()+setKafkaConfValue (KafkaConf confPtr) key value = do+ allocaBytes nErrorBytes $ \charPtr -> do+ err <- rdKafkaConfSet confPtr key value charPtr (fromIntegral nErrorBytes)+ checkConfSetValue err charPtr++setKafkaTopicConfValue :: KafkaTopicConf -> String -> String -> IO ()+setKafkaTopicConfValue (KafkaTopicConf confPtr) key value = do+ allocaBytes nErrorBytes $ \charPtr -> do+ err <- rdKafkaTopicConfSet confPtr key value charPtr (fromIntegral nErrorBytes)+ checkConfSetValue err charPtr++setKafkaLogLevel :: Kafka -> KafkaLogLevel -> IO ()+setKafkaLogLevel (Kafka kptr _) level = + rdKafkaSetLogLevel kptr (fromEnum level)+ newKafka :: KafkaType -> KafkaConf -> IO Kafka-newKafka kafkaType (KafkaConf confPtr) = do+newKafka kafkaType c@(KafkaConf confPtr) = do et <- newRdKafkaT (kafkaTypeToRdKafkaType kafkaType) confPtr case et of Left e -> error e- Right x -> return $ Kafka x+ Right x -> return $ Kafka x c newKafkaTopic :: Kafka -> String -> KafkaTopicConf -> IO (KafkaTopic)-newKafkaTopic k@(Kafka kafkaPtr) topicName (KafkaTopicConf confPtr) = do- et <- newRdKafkaTopicT kafkaPtr topicName confPtr+newKafkaTopic k@(Kafka kPtr _) tName conf@(KafkaTopicConf confPtr) = do+ et <- newRdKafkaTopicT kPtr tName confPtr case et of Left e -> throw $ KafkaError e- Right x -> return $ KafkaTopic x k+ Right x -> return $ KafkaTopic x k conf addBrokers :: Kafka -> String -> IO ()-addBrokers (Kafka kptr) brokerStr = do+addBrokers (Kafka kptr _) brokerStr = do numBrokers <- rdKafkaBrokersAdd kptr brokerStr when (numBrokers == 0) (throw $ KafkaBadSpecification "No valid brokers specified") startConsuming :: KafkaTopic -> Int -> KafkaOffset -> IO ()-startConsuming (KafkaTopic topicPtr _) partition offset = +startConsuming (KafkaTopic topicPtr _ _) partition offset = let trueOffset = case offset of- KafkaOffsetBeginning -> -2- KafkaOffsetEnd -> -1- KafkaOffsetStored -> -1000+ KafkaOffsetBeginning -> (- 2)+ KafkaOffsetEnd -> (- 1)+ KafkaOffsetStored -> (- 1000) KafkaOffset i -> i in throwOnError $ rdKafkaConsumeStart topicPtr partition trueOffset @@ -147,29 +222,64 @@ BSI.memcpy bsptr ptr len +fromMessageStorable :: RdKafkaMessageT -> IO (KafkaMessage)+fromMessageStorable s = do+ payload <- word8PtrToBS (len'RdKafkaMessageT s) (payload'RdKafkaMessageT s) ++ key <- if (key'RdKafkaMessageT s) == nullPtr then return Nothing + else word8PtrToBS (keyLen'RdKafkaMessageT s) (key'RdKafkaMessageT s) >>= return . Just++ return $ KafkaMessage+ (partition'RdKafkaMessageT s) + (offset'RdKafkaMessageT s)+ payload+ key + consumeMessage :: KafkaTopic -> Int -> Int -> IO (Either KafkaError KafkaMessage)-consumeMessage (KafkaTopic topicPtr _) partition timeout = do- ptr <- rdKafkaConsume topicPtr (fromIntegral partition) (fromIntegral timeout)- withForeignPtr ptr $ \realPtr ->- if realPtr == nullPtr then getErrno >>= return . Left . kafkaRespErr +consumeMessage (KafkaTopic topicPtr _ _) partition timeout = do+ ptr <- rdKafkaConsume topicPtr (fromIntegral partition) (fromIntegral timeout)+ withForeignPtr ptr $ \realPtr ->+ if realPtr == nullPtr then getErrno >>= return . Left . kafkaRespErr + else do+ addForeignPtrFinalizer rdKafkaMessageDestroy ptr+ s <- peek realPtr+ if (err'RdKafkaMessageT s) /= RdKafkaRespErrNoError then return $ Left $ KafkaResponseError $ err'RdKafkaMessageT s else do- addForeignPtrFinalizer rdKafkaMessageDestroy ptr- s <- peek realPtr- if (err'RdKafkaMessageT s) /= RdKafkaRespErrNoError then return $ Left $ KafkaResponseError $ err'RdKafkaMessageT s- else do- payload <- word8PtrToBS (len'RdKafkaMessageT s) (payload'RdKafkaMessageT s) + payload <- word8PtrToBS (len'RdKafkaMessageT s) (payload'RdKafkaMessageT s) - key <- if (key'RdKafkaMessageT s) == nullPtr then return Nothing - else word8PtrToBS (keyLen'RdKafkaMessageT s) (key'RdKafkaMessageT s) >>= return . Just+ key <- if (key'RdKafkaMessageT s) == nullPtr then return Nothing + else word8PtrToBS (keyLen'RdKafkaMessageT s) (key'RdKafkaMessageT s) >>= return . Just - return $ Right $ KafkaMessage- (partition'RdKafkaMessageT s) - (offset'RdKafkaMessageT s)- payload- key + return $ Right $ KafkaMessage+ (partition'RdKafkaMessageT s) + (offset'RdKafkaMessageT s)+ payload+ key +consumeMessageBatch :: KafkaTopic -> Int -> Int -> Int -> IO (Either KafkaError [KafkaMessage])+consumeMessageBatch (KafkaTopic topicPtr _ _) partition timeout maxMessages = + allocaArray maxMessages $ \outputPtr -> do+ numMessages <- rdKafkaConsumeBatch topicPtr (fromIntegral partition) timeout outputPtr (fromIntegral maxMessages)+ if numMessages < 0 then getErrno >>= return . Left . kafkaRespErr+ else do+ ms <- forM [0..(numMessages - 1)] $ \mnum -> do + storablePtr <- peekElemOff outputPtr (fromIntegral mnum)+ storable <- peek storablePtr+ ret <- fromMessageStorable storable+ fptr <- newForeignPtr_ storablePtr+ addForeignPtrFinalizer rdKafkaMessageDestroy fptr+ return ret+ return $ Right ms ++storeOffset :: KafkaTopic -> Int -> Int -> IO (Maybe KafkaError)+storeOffset (KafkaTopic topicPtr _ _) partition offset = do+ err <- rdKafkaOffsetStore topicPtr (fromIntegral partition) (fromIntegral offset)+ case err of + RdKafkaRespErrNoError -> return Nothing+ e -> return $ Just $ KafkaResponseError e+ produceMessage :: KafkaTopic -> KafkaProducePartition -> KafkaProduceMessage -> IO (Maybe KafkaError)-produceMessage (KafkaTopic topicPtr _) partition (KafkaProduceMessage payload) = do+produceMessage (KafkaTopic topicPtr _ _) partition (KafkaProduceMessage payload) = do let (payloadFPtr, payloadOffset, payloadLength) = BSI.toForeignPtr payload withForeignPtr payloadFPtr $ \payloadPtr -> do@@ -180,7 +290,11 @@ copyMsgFlags passedPayload (fromIntegral payloadLength) nullPtr (CSize 0) nullPtr -produceMessage (KafkaTopic topicPtr _) partition (KafkaProduceKeyedMessage key payload) = do+produceMessage _ _ (KafkaProduceKeyedMessage _ _) = undefined++produceKeyedMessage :: KafkaTopic -> KafkaProduceMessage -> IO (Maybe KafkaError)+produceKeyedMessage _ (KafkaProduceMessage _) = undefined+produceKeyedMessage (KafkaTopic topicPtr _ _) (KafkaProduceKeyedMessage key payload) = do let (payloadFPtr, payloadOffset, payloadLength) = BSI.toForeignPtr payload (keyFPtr, keyOffset, keyLength) = BSI.toForeignPtr key @@ -190,12 +304,12 @@ passedKey = keyPtr `plusPtr` keyOffset handleProduceErr =<< - rdKafkaProduce topicPtr (producePartitionInteger partition)+ rdKafkaProduce topicPtr (producePartitionInteger KafkaUnassignedPartition) copyMsgFlags passedPayload (fromIntegral payloadLength) passedKey (fromIntegral keyLength) nullPtr produceMessageBatch :: KafkaTopic -> KafkaProducePartition -> [KafkaProduceMessage] -> IO ([(KafkaProduceMessage, KafkaError)])-produceMessageBatch (KafkaTopic topicPtr _) partition pms = do+produceMessageBatch (KafkaTopic topicPtr _ _) partition pms = do storables <- forM pms produceMessageToMessage withArray storables $ \batchPtr -> do batchPtrF <- newForeignPtr_ batchPtr@@ -220,7 +334,7 @@ , keyLen'RdKafkaMessageT = 0 , key'RdKafkaMessageT = nullPtr }- produceMessageToMessage (KafkaProduceKeyedMessage bs kbs) = do+ produceMessageToMessage (KafkaProduceKeyedMessage kbs bs) = do let (payloadFPtr, payloadOffset, payloadLength) = BSI.toForeignPtr bs (keyFPtr, keyOffset, keyLength) = BSI.toForeignPtr kbs @@ -236,11 +350,49 @@ , payload'RdKafkaMessageT = passedPayload , offset'RdKafkaMessageT = 0 , keyLen'RdKafkaMessageT = keyLength- , key'RdKafkaMessageT = keyPtr+ , key'RdKafkaMessageT = passedKey }- - +type ConfigOverrides = [(String, String)]+withKafkaProducer :: ConfigOverrides -> ConfigOverrides + -> String -> String + -> (Kafka -> KafkaTopic -> IO a) + -> IO a+withKafkaProducer configOverrides topicConfigOverrides brokerString tName cb =+ bracket + (do+ conf <- newKafkaConf+ mapM_ (\(k, v) -> setKafkaConfValue conf k v) configOverrides+ kafka <- newKafka KafkaProducer conf+ addBrokers kafka brokerString+ topicConf <- newKafkaTopicConf+ mapM_ (\(k, v) -> setKafkaTopicConfValue topicConf k v) topicConfigOverrides + topic <- newKafkaTopic kafka tName topicConf+ return (kafka, topic)+ )+ (\(kafka, _) -> drainOutQueue kafka)+ (\(k, t) -> cb k t)++withKafkaConsumer :: ConfigOverrides -> ConfigOverrides + -> String -> String -> Int -> KafkaOffset + -> (Kafka -> KafkaTopic -> IO a) + -> IO a+withKafkaConsumer configOverrides topicConfigOverrides brokerString tName partition offset cb =+ bracket+ (do+ conf <- newKafkaConf+ mapM_ (\(k, v) -> setKafkaConfValue conf k v) configOverrides+ kafka <- newKafka KafkaConsumer conf+ addBrokers kafka brokerString+ topicConf <- newKafkaTopicConf+ mapM_ (\(k, v) -> setKafkaTopicConfValue topicConf k v) topicConfigOverrides + topic <- newKafkaTopic kafka tName topicConf+ startConsuming topic partition offset+ return (kafka, topic)+ )+ (\(_, topic) -> stopConsuming topic partition)+ (\(k, t) -> cb k t)+ {-# INLINE copyMsgFlags #-} copyMsgFlags :: Int copyMsgFlags = rdKafkaMsgFlagCopy@@ -280,6 +432,7 @@ , partitionIsrs :: [Int] } deriving (Eq, Show, Typeable) + getAllMetadata :: Kafka -> Int -> IO (Either KafkaError KafkaMetadata) getAllMetadata k timeout = getMetadata k Nothing timeout @@ -294,9 +447,9 @@ _ -> return $ Left $ KafkaError "Incorrect number of topics returned" getMetadata :: Kafka -> Maybe KafkaTopic -> Int -> IO (Either KafkaError KafkaMetadata)-getMetadata (Kafka kPtr) mTopic timeout = alloca $ \mdDblPtr -> do+getMetadata (Kafka kPtr _) mTopic timeout = alloca $ \mdDblPtr -> do err <- case mTopic of - Just (KafkaTopic kTopicPtr _) -> + Just (KafkaTopic kTopicPtr _ _) -> rdKafkaMetadata kPtr False kTopicPtr mdDblPtr timeout Nothing -> do nullTopic <- newForeignPtr_ nullPtr@@ -357,10 +510,10 @@ e -> return $ Left $ KafkaResponseError e pollEvents :: Kafka -> Int -> IO ()-pollEvents (Kafka kPtr) timeout = rdKafkaPoll kPtr timeout >> return ()+pollEvents (Kafka kPtr _) timeout = rdKafkaPoll kPtr timeout >> return () outboundQueueLength :: Kafka -> IO (Int)-outboundQueueLength (Kafka kPtr) = rdKafkaOutqLen kPtr+outboundQueueLength (Kafka kPtr _) = rdKafkaOutqLen kPtr drainOutQueue :: Kafka -> IO () drainOutQueue k = do@@ -373,7 +526,7 @@ kafkaRespErr (Errno num) = KafkaResponseError $ rdKafkaErrno2err (fromIntegral num) stopConsuming :: KafkaTopic -> Int -> IO ()-stopConsuming (KafkaTopic topicPtr _) partition = +stopConsuming (KafkaTopic topicPtr _ _) partition = throwOnError $ rdKafkaConsumeStop topicPtr partition throwOnError :: IO (Maybe String) -> IO ()@@ -391,6 +544,13 @@ dumpKafkaConf :: KafkaConf -> IO (Map String String) dumpKafkaConf (KafkaConf kptr) = do parseDump (\sizeptr -> rdKafkaConfDump kptr sizeptr)++dumpConfFromKafka :: Kafka -> IO (Map String String) +dumpConfFromKafka (Kafka _ cfg) = dumpKafkaConf cfg++dumpConfFromKafkaTopic :: KafkaTopic -> IO (Map String String)+dumpConfFromKafkaTopic (KafkaTopic _ _ conf) = dumpKafkaTopicConf conf+ parseDump :: (CSizePtr -> IO (Ptr CString)) -> IO (Map String String) parseDump cstr = alloca $ \sizeptr -> do
+ src/Haskakafka/Example.hs view
@@ -0,0 +1,64 @@+module Haskakafka.Example where++import Haskakafka++import qualified Data.ByteString.Char8 as C8++example :: IO ()+example = do+ let + -- Optionally, we can configure certain parameters for Kafka+ kafkaConfig = [("socket.timeout.ms", "50000")]+ topicConfig = [("request.timeout.ms", "50000")]++ -- Payloads are just ByteStrings+ samplePayload = C8.pack "Hello world"+++ -- withKafkaProducer opens a producer connection and gives us + -- two objects for subsequent use.+ withKafkaProducer kafkaConfig topicConfig + "localhost:9092" "test_topic" + $ \kafka topic -> do++ -- Produce a single unkeyed message to partition 0+ let message = KafkaProduceMessage samplePayload+ _ <- produceMessage topic (KafkaSpecifiedPartition 0) message++ -- Produce a single keyed message+ let keyMessage = KafkaProduceKeyedMessage (C8.pack "Key") samplePayload+ _ <- produceKeyedMessage topic message++ -- We can also use the batch API for better performance+ _ <- produceMessageBatch topic KafkaUnassignedPartition [message, keyMessage]++ putStrLn "Done producing messages, here was our config: "+ dumpConfFromKafka kafka >>= \d -> putStrLn $ "Kafka config: " ++ (show d)+ dumpConfFromKafkaTopic topic >>= \d -> putStrLn $ "Topic config: " ++ (show d)+++ -- withKafkaConsumer opens a consumer connection and starts consuming+ let partition = 0+ withKafkaConsumer kafkaConfig topicConfig + "localhost:9092" "test_topic"+ partition -- locked to a specific partition for each consumer+ KafkaOffsetBeginning -- start reading from beginning (alternatively, use+ -- KafkaOffsetEnd, KafkaOffset or KafkaOffsetStored)+ $ \kafka topic -> do+ -- Consume a single message at a time+ let timeoutMs = 1000+ me <- consumeMessage topic partition timeoutMs+ case me of + (Left err) -> putStrLn $ "Uh oh, an error! " ++ (show err)+ (Right m) -> putStrLn $ "Woo, payload was " ++ (C8.unpack $ messagePayload m)++ -- For better performance, consume in batches+ let maxMessages = 10+ mes <- consumeMessageBatch topic partition timeoutMs maxMessages+ case mes of + (Left err) -> putStrLn $ "Something went wrong in batch consume! " ++ (show err)+ (Right ms) -> putStrLn $ "Woohoo, we got " ++ (show $ length ms) ++ " messages"+++ -- Be a little less noisy+ setKafkaLogLevel kafka KafkaLogCrit
src/Haskakafka/Internal.chs view
@@ -6,7 +6,7 @@ import Control.Applicative import Control.Monad import Data.Word-import Foreign+import Foreign hiding (unsafePerformIO) import Foreign.C.Error import Foreign.C.String import Foreign.C.Types@@ -16,8 +16,6 @@ import System.Posix.IO import System.Posix.Types -import qualified Data.Map.Strict as Map- #include "rdkafka.h" type CInt64T = {#type int64_t #}@@ -35,6 +33,10 @@ rdKafkaMsgFlagCopy :: RdKafkaMsgFlag rdKafkaMsgFlagCopy = 0x2 +-- Number of bytes allocated for an error buffer+nErrorBytes :: Int+nErrorBytes = 1024 * 8+ -- Helper functions {#fun pure unsafe rd_kafka_version as ^ {} -> `Int' #}@@ -184,6 +186,10 @@ {#pointer *rd_kafka_metadata_t as RdKafkaMetadataTPtr foreign -> RdKafkaMetadataT #} +-- rd_kafka_message+foreign import ccall unsafe "rdkafka.h &rd_kafka_message_destroy"+ rdKafkaMessageDestroy :: FunPtr (Ptr RdKafkaMessageT -> IO ())+ -- rd_kafka_conf {#fun unsafe rd_kafka_conf_new as ^ {} -> `RdKafkaConfTPtr' #}@@ -194,6 +200,10 @@ {#fun unsafe rd_kafka_conf_dup as ^ {`RdKafkaConfTPtr'} -> `RdKafkaConfTPtr' #} +{#fun unsafe rd_kafka_conf_set as ^+ {`RdKafkaConfTPtr', `String', `String', id `CCharBufPointer', cIntConv `CSize'} + -> `RdKafkaConfResT' cIntToEnum #}+ newRdKafkaConfT :: IO RdKafkaConfTPtr newRdKafkaConfT = do ret <- rdKafkaConfNew@@ -218,6 +228,10 @@ foreign import ccall unsafe "rdkafka.h &rd_kafka_topic_conf_destroy" rdKafkaTopicConfDestroy :: FunPtr (Ptr RdKafkaTopicConfT -> IO ()) +{#fun unsafe rd_kafka_topic_conf_set as ^+ {`RdKafkaTopicConfTPtr', `String', `String', id `CCharBufPointer', cIntConv `CSize'} + -> `RdKafkaConfResT' cIntToEnum #}+ newRdKafkaTopicConfT :: IO RdKafkaTopicConfTPtr newRdKafkaTopicConfT = do ret <- rdKafkaTopicConfNew@@ -235,9 +249,6 @@ foreign import ccall unsafe "rdkafka.h &rd_kafka_destroy" rdKafkaDestroy :: FunPtr (Ptr RdKafkaT -> IO ()) -nErrorBytes :: Int-nErrorBytes = 1024 * 8- newRdKafkaT :: RdKafkaTypeT -> RdKafkaConfTPtr -> IO (Either String RdKafkaTPtr) newRdKafkaT kafkaType confPtr = allocaBytes nErrorBytes $ \charPtr -> do@@ -252,6 +263,11 @@ {#fun unsafe rd_kafka_brokers_add as ^ {`RdKafkaTPtr', `String'} -> `Int' #} +{#fun unsafe rd_kafka_set_log_level as ^+ {`RdKafkaTPtr', `Int'} -> `()' #}++-- rd_kafka consume+ {#fun unsafe rd_kafka_consume_start as rdKafkaConsumeStartInternal {`RdKafkaTopicTPtr', cIntConv `CInt32T', cIntConv `CInt64T'} -> `Int' #} @@ -265,10 +281,11 @@ {`RdKafkaTopicTPtr', cIntConv `CInt32T'} -> `Int' #} {#fun unsafe rd_kafka_consume as ^- {`RdKafkaTopicTPtr', cIntConv `CInt32T', `Int'} -> `RdKafkaMessageTPtr' #}+ {`RdKafkaTopicTPtr', cIntConv `CInt32T', `Int'} -> `RdKafkaMessageTPtr' #} -foreign import ccall unsafe "rdkafka.h &rd_kafka_message_destroy"- rdKafkaMessageDestroy :: FunPtr (Ptr RdKafkaMessageT -> IO ())+{#fun unsafe rd_kafka_consume_batch as ^+ {`RdKafkaTopicTPtr', cIntConv `CInt32T', `Int', castPtr `Ptr (Ptr RdKafkaMessageT)', cIntConv `CSize'}+ -> `CSize' cIntConv #} rdKafkaConsumeStop :: RdKafkaTopicTPtr -> Int -> IO (Maybe String) rdKafkaConsumeStop topicPtr partition = do@@ -277,7 +294,12 @@ -1 -> kafkaErrnoString >>= return . Just _ -> return Nothing +{#fun unsafe rd_kafka_offset_store as rdKafkaOffsetStore+ {`RdKafkaTopicTPtr', cIntConv `CInt32T', cIntConv `CInt64T'} + -> `RdKafkaRespErrT' cIntToEnum #} +-- rd_kafka produce+ {#fun unsafe rd_kafka_produce as ^ {`RdKafkaTopicTPtr', cIntConv `CInt32T', `Int', castPtr `Word8Ptr', cIntConv `CSize', castPtr `Word8Ptr', cIntConv `CSize', castPtr `Word8Ptr'}@@ -289,6 +311,8 @@ castMetadata :: Ptr (Ptr RdKafkaMetadataT) -> Ptr (Ptr ()) castMetadata ptr = castPtr ptr +-- rd_kafka_metadata+ {#fun unsafe rd_kafka_metadata as ^ {`RdKafkaTPtr', boolToCInt `Bool', `RdKafkaTopicTPtr', castMetadata `Ptr (Ptr RdKafkaMetadataT)', `Int'}@@ -323,7 +347,6 @@ else do addForeignPtrFinalizer rdKafkaTopicDestroy ret return $ Right ret- -- Marshall / Unmarshall enumToCInt :: Enum a => a -> CInt
tests/TestMain.hs view
@@ -1,64 +1,185 @@-module Main (testmain, main) where-+{-# LANGUAGE ScopedTypeVariables #-} +module Main (main) where import Haskakafka +import Control.Exception import Control.Monad+import System.Environment import Test.Hspec-import Control.Exception (evaluate)+import Text.Regex.Posix -import qualified Data.ByteString.Char8 as BS+import qualified Data.Map as Map+import qualified Data.ByteString.Char8 as C8 +brokerAddress :: IO String+brokerAddress = (getEnv "HASKAKAFKA_TEST_BROKER") `catch` \(_ :: SomeException) -> (return "localhost:9092")+brokerTopic :: IO String+brokerTopic = (getEnv "HASKAKAFKA_TEST_TOPIC") `catch` \(_ :: SomeException) -> (return "haskakafka_tests")+kafkaDelay :: Int -- Little delay for operation+kafkaDelay = 5 * 1000++getAddressTopic :: (String -> String -> IO ()) -> IO ()+getAddressTopic cb = do+ b <- brokerAddress+ t <- brokerTopic+ cb b t++sampleProduceMessages :: [KafkaProduceMessage]+sampleProduceMessages = + [ (KafkaProduceMessage $ C8.pack "hello")+ , (KafkaProduceKeyedMessage (C8.pack "key") (C8.pack "value"))+ , (KafkaProduceMessage $ C8.pack "goodbye")+ ]++shouldBeProduceConsume :: KafkaProduceMessage -> KafkaMessage -> IO ()+shouldBeProduceConsume (KafkaProduceMessage ppayload) m = do+ (messagePayload m) `shouldBe` ppayload+ (messageKey m) `shouldBe` Nothing++shouldBeProduceConsume (KafkaProduceKeyedMessage pkey ppayload) m = do+ ppayload `shouldBe` (messagePayload m)+ (Just pkey) `shouldBe` (messageKey m)++shouldBeEmptyTopic :: KafkaTopic -> IO ()+shouldBeEmptyTopic kt = do+ eof <- consumeMessage kt 0 kafkaDelay+ eof `shouldBe` (Left $ KafkaResponseError $ RdKafkaRespErrPartitionEof)+ testmain :: IO () testmain = hspec $ do- describe "Prelude.head" $ do- it "returns the first element of a list" $ do- head [23 ..] `shouldBe` (23 :: Int)+ describe "RdKafka versioning" $ do+ it "should be a valid version number" $ do+ rdKafkaVersionStr `shouldSatisfy` (=~"[0-9]+(.[0-9]+)+") - it "throws an exception if used with an empty list" $ do- evaluate (head []) `shouldThrow` anyException+ describe "Supported properties" $ do+ it "should list supported properties" $ do+ props <- supportedKafkaConfProperties+ props `shouldSatisfy` (\x -> (length x) > 0) -doConsume :: IO ()-doConsume = do- kConf <- newKafkaConf- kafka <- newKafka KafkaConsumer kConf- addBrokers kafka "localhost:9092"- kTopicConf <- newKafkaTopicConf- topic <- newKafkaTopic kafka "test" kTopicConf+ describe "Kafka Configuration" $ do+ it "should allow dumping" $ do+ kConf <- newKafkaConf + kvs <- dumpKafkaConf kConf+ (Map.size kvs) `shouldSatisfy` (>0) - startConsuming topic 0 (KafkaOffsetBeginning)- _ <- forever $ do- m <- consumeMessage topic 0 (1000 * 1000)- print m- stopConsuming topic 0+ it "should change when set is called" $ do+ kConf <- newKafkaConf + setKafkaConfValue kConf "socket.timeout.ms" "50000"+ kvs <- dumpKafkaConf kConf+ (kvs Map.! "socket.timeout.ms") `shouldBe` "50000" -doProduce :: IO ()-doProduce = do- kConf <- newKafkaConf- kafka <- newKafka KafkaProducer kConf- addBrokers kafka "localhost:9092"- kTopicConf <- newKafkaTopicConf- topic <- newKafkaTopic kafka "test" kTopicConf- let me = KafkaMessage 0 0 (BS.pack "hi") Nothing- err <- produceMessage topic me+ it "should throw an exception on unknown property" $ do+ kConf <- newKafkaConf+ (setKafkaConfValue kConf "blippity.blop.cosby" "120") `shouldThrow`+ (\(KafkaUnknownConfigurationKey str) -> (length str) > 0) - drainOutQueue kafka- - print err+ it "should throw an exception on an invalid value" $ do+ kConf <- newKafkaConf+ (setKafkaConfValue kConf "socket.timeout.ms" "monorail") `shouldThrow`+ (\(KafkaInvalidConfigurationValue str) -> (length str) > 0)+ + describe "Kafka topic configuration" $ do+ it "should allow dumping" $ do+ kConf <- newKafkaTopicConf+ kvs <- dumpKafkaTopicConf kConf+ (Map.size kvs) `shouldSatisfy` (>0) + it "should change when set is called" $ do+ kConf <- newKafkaTopicConf+ setKafkaTopicConfValue kConf "request.timeout.ms" "20000"+ kvs <- dumpKafkaTopicConf kConf+ (kvs Map.! "request.timeout.ms") `shouldBe` "20000" -main :: IO ()-main = do- doProduce- doConsume+ it "should throw an exception on unknown property" $ do+ kConf <- newKafkaTopicConf+ (setKafkaTopicConfValue kConf "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`+ (\(KafkaInvalidConfigurationValue str) -> (length str) > 0) - -- hPrintKafkaProperties stdout- --o <- c_stdout- --rdKafkaConfPropertiesShow o- --putStrLn $ "Kafka version: " ++ rdKafkaVersionStr- --t <- c_rd_kafka_topic_conf_new- --topics <- dumpTopics t- --print topics- --c_rd_kafka_topic_conf_destroy t- --putStrLn $ "Cleaned up"- + describe "Logging" $ do+ it "should allow setting of log level" $ getAddressTopic $ \a t -> do+ withKafkaConsumer [] [] a t 0 KafkaOffsetEnd $ \kafka _ -> do+ setKafkaLogLevel kafka KafkaLogDebug ++ describe "Consume and produce cycle" $ do+ it "should be able to produce and consume a unkeyed message off of the broker" $ getAddressTopic $ \a t -> do+ let message = KafkaProduceMessage (C8.pack "hey hey we're the monkeys")+ withKafkaConsumer [] [] a t 0 KafkaOffsetEnd $ \_ topic -> do+ eof <- consumeMessage topic 0 kafkaDelay+ eof `shouldBe` (Left $ KafkaResponseError $ RdKafkaRespErrPartitionEof)+ perr <- withKafkaProducer [] [] a t $ \_ producerTopic -> do + produceMessage producerTopic (KafkaSpecifiedPartition 0) message+ perr `shouldBe` Nothing+ + et <- consumeMessage topic 0 kafkaDelay+ case et of + Left err -> error $ show err+ Right m -> message `shouldBeProduceConsume` m++ it "should be able to produce and consume a keyed message" $ getAddressTopic $ \a t -> do+ let message = KafkaProduceKeyedMessage (C8.pack "key") (C8.pack "monkey around")++ withKafkaConsumer [] [] a t 0 KafkaOffsetEnd $ \_ topic -> do+ shouldBeEmptyTopic topic+ perr <- withKafkaProducer [] [] a t $ \_ producerTopic -> do+ produceKeyedMessage producerTopic message+ perr `shouldBe` Nothing++ et <- consumeMessage topic 0 kafkaDelay+ case et of+ Left err -> error $ show err+ Right m -> message `shouldBeProduceConsume` m++ it "should be able to batch produce messages" $ getAddressTopic $ \a t -> do+ withKafkaConsumer [] [] a t 0 KafkaOffsetEnd $ \_ topic -> do+ shouldBeEmptyTopic topic+ errs <- withKafkaProducer [] [] a t $ \_ producerTopic -> do+ produceMessageBatch producerTopic (KafkaSpecifiedPartition 0 ) sampleProduceMessages+ errs `shouldBe` []++ ets <- mapM (\_ -> consumeMessage topic 0 kafkaDelay) ([1..3] :: [Integer])++ forM_ (zip sampleProduceMessages ets) $ \(pm, et) -> + case (pm, et) of+ (_, Left err) -> error $ show err+ (pmessage, Right cm) -> pmessage `shouldBeProduceConsume` cm++ it "should be able to batch consume messages" $ getAddressTopic $ \a t -> do+ withKafkaConsumer [] [] a t 0 KafkaOffsetEnd $ \_ topic -> do+ shouldBeEmptyTopic topic+ errs <- withKafkaProducer [] [] a t $ \_ producerTopic -> do+ produceMessageBatch producerTopic (KafkaSpecifiedPartition 0 ) sampleProduceMessages+ errs `shouldBe` []++ et <- consumeMessageBatch topic 0 (1000) 3+ case et of + (Left err) -> error $ show err+ (Right oms) -> do+ (length oms) `shouldBe` 3+ forM_ (zip sampleProduceMessages oms) $ \(pm, om) -> pm `shouldBeProduceConsume` om++-- Test setup (error on no Kafka)+checkForKafka :: IO (Bool)+checkForKafka = do+ kConf <- newKafkaConf + kafka <- newKafka KafkaConsumer kConf+ a <- brokerAddress+ addBrokers kafka a+ me <- getAllMetadata kafka 1000+ return $ case me of + (Left _) -> False+ (Right _) -> True++main :: IO () +main = do + hasKafka <- checkForKafka + if hasKafka then testmain+ else error "\n\n\+ \*******************************************************************************\n\+ \*Haskakafka's tests require an operable Kafka broker running on localhost:9092*\n\+ \*please follow the guide in Readme.md to set this up *\n\+ \*******************************************************************************\n"