diff --git a/haskakafka.cabal b/haskakafka.cabal --- a/haskakafka.cabal +++ b/haskakafka.cabal @@ -1,17 +1,16 @@ name: haskakafka -version: 0.5.0 +version: 1.0.0 synopsis: Kafka bindings for Haskell -description: Use Apache Kafka in Haskell through the librdkafka - C library. +description: Apache Kafka bindings backed by the librdkafka + C library. This implies full consumer and producer + support for Kafka 0.8.x. homepage: http://github.com/cosbynator/haskakafka license: MIT license-file: LICENSE author: Thomas Dimson maintainer: Thomas Dimson --- copyright: category: Network build-type: Simple --- extra-source-files: cabal-version: >=1.10 source-repository head @@ -27,9 +26,12 @@ , unix exposed-modules: Haskakafka - Haskakafka.Internal - Haskakafka.InternalEnum + Haskakafka.InternalRdKafka + Haskakafka.InternalRdKafkaEnum + Haskakafka.InternalSetup + Haskakafka.InternalTypes Haskakafka.Example + other-modules: hs-source-dirs: src default-language: Haskell2010 ghc-options: -Wall diff --git a/src/Haskakafka.hs b/src/Haskakafka.hs --- a/src/Haskakafka.hs +++ b/src/Haskakafka.hs @@ -1,212 +1,83 @@ -{-# LANGUAGE DeriveDataTypeable #-} - module Haskakafka -( Kafka - , KafkaConf - , KafkaLogLevel(..) - , KafkaMessage(..) - , KafkaOffset(..) - , KafkaMetadata(..) - , KafkaBrokerMetadata(..) - , KafkaTopicMetadata(..) - , KafkaPartitionMetadata(..) - , KafkaProduceMessage(..) - , KafkaProducePartition(..) - , KafkaTopic - , KafkaTopicConf - , KafkaType(..) - , KafkaError (..) - , rdKafkaVersionStr - , supportedKafkaConfProperties - , hPrintKafkaProperties - , hPrintKafka - , newKafkaConf - , setKafkaConfValue - , setKafkaTopicConfValue - , setKafkaLogLevel - , dumpKafkaConf - , newKafkaTopicConf - , newKafka - , newKafkaTopic - , dumpKafkaTopicConf - , addBrokers - , startConsuming - , consumeMessage - , consumeMessageBatch - , storeOffset - , stopConsuming - , produceMessage - , produceKeyedMessage - , produceMessageBatch - , pollEvents - , drainOutQueue - , getAllMetadata - , getTopicMetadata - , withKafkaProducer - , withKafkaConsumer - , dumpConfFromKafkaTopic - , dumpConfFromKafka - , module Haskakafka.InternalEnum -) where - -import Control.Exception -import Control.Monad -import Data.Map.Strict (Map) -import Data.Typeable -import Foreign -import Foreign.C.Error -import Foreign.C.String -import Foreign.C.Types -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 - -data KafkaMessage = KafkaMessage - { messagePartition :: !Int - , messageOffset :: !Int64 - , messagePayload :: !BS.ByteString - , messageKey :: Maybe BS.ByteString - } - deriving (Eq, Show, Read, Typeable) - -data KafkaProduceMessage = - KafkaProduceMessage - {-# UNPACK #-} !BS.ByteString - | KafkaProduceKeyedMessage - {-# UNPACK #-} !BS.ByteString -- | Key - {-# UNPACK #-} !BS.ByteString -- | Payload - deriving (Eq, Show, Typeable) - - -data KafkaProducePartition = - KafkaSpecifiedPartition {-# UNPACK #-} !Int - | KafkaUnassignedPartition - -data KafkaOffset = KafkaOffsetBeginning - | KafkaOffsetEnd - | KafkaOffsetStored - | KafkaOffset Int64 - -data KafkaConf = KafkaConf RdKafkaConfTPtr -data KafkaTopicConf = KafkaTopicConf RdKafkaTopicConfTPtr - -data KafkaType = KafkaConsumer | KafkaProducer -data Kafka = Kafka { kafkaPtr :: RdKafkaTPtr, _kafkaConf :: KafkaConf} -data KafkaTopic = KafkaTopic - RdKafkaTopicTPtr - Kafka -- Kept around to prevent garbage collection - KafkaTopicConf - - -kafkaTypeToRdKafkaType :: KafkaType -> RdKafkaTypeT -kafkaTypeToRdKafkaType KafkaConsumer = RdKafkaConsumer -kafkaTypeToRdKafkaType KafkaProducer = RdKafkaProducer +( fetchBrokerMetadata +, withKafkaConsumer +, consumeMessage +, consumeMessageBatch +, withKafkaProducer +, produceMessage +, produceKeyedMessage +, produceMessageBatch +, storeOffset +, getAllMetadata +, getTopicMetadata --- 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 +-- Internal objects +, IS.newKafka +, IS.newKafkaTopic +, IS.dumpConfFromKafka +, IS.dumpConfFromKafkaTopic +, IS.setLogLevel +, IS.hPrintSupportedKafkaConf +, IS.hPrintKafka +, rdKafkaVersionStr -hPrintKafkaProperties :: Handle -> IO () -hPrintKafkaProperties h = handleToCFile h "w" >>= rdKafkaConfPropertiesShow +-- Type re-exports +, IT.Kafka +, IT.KafkaTopic -hPrintKafka :: Handle -> Kafka -> IO () -hPrintKafka h k = handleToCFile h "w" >>= \f -> rdKafkaDump f (kafkaPtr k) +, IT.KafkaOffset(..) +, IT.KafkaMessage(..) -newKafkaTopicConf :: IO KafkaTopicConf -newKafkaTopicConf = newRdKafkaTopicConfT >>= return . KafkaTopicConf +, IT.KafkaProduceMessage(..) +, IT.KafkaProducePartition(..) -newKafkaConf :: IO KafkaConf -newKafkaConf = newRdKafkaConfT >>= return . KafkaConf +, IT.KafkaMetadata(..) +, IT.KafkaBrokerMetadata(..) +, IT.KafkaTopicMetadata(..) +, IT.KafkaPartitionMetadata(..) -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 +, IT.KafkaLogLevel(..) +, IT.KafkaError(..) +, RDE.RdKafkaRespErrT -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 +-- Pseudo-internal +, addBrokers +, startConsuming +, stopConsuming +, drainOutQueue -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 +) where -setKafkaLogLevel :: Kafka -> KafkaLogLevel -> IO () -setKafkaLogLevel (Kafka kptr _) level = - rdKafkaSetLogLevel kptr (fromEnum level) +import Haskakafka.InternalRdKafka +import Haskakafka.InternalRdKafkaEnum +import Haskakafka.InternalSetup +import Haskakafka.InternalTypes -newKafka :: KafkaType -> KafkaConf -> IO Kafka -newKafka kafkaType c@(KafkaConf confPtr) = do - et <- newRdKafkaT (kafkaTypeToRdKafkaType kafkaType) confPtr - case et of - Left e -> error e - Right x -> return $ Kafka x c +import Control.Exception +import Control.Monad +import Foreign +import Foreign.C.Error +import Foreign.C.String +import Foreign.C.Types -newKafkaTopic :: Kafka -> String -> KafkaTopicConf -> IO (KafkaTopic) -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 conf +import qualified Data.ByteString as BS +import qualified Data.ByteString.Internal as BSI +import qualified Haskakafka.InternalTypes as IT +import qualified Haskakafka.InternalSetup as IS +import qualified Haskakafka.InternalRdKafkaEnum as RDE +-- | Adds a broker string to a given kafka instance. You +-- probably shouldn't use this directly (see 'withKafkaConsumer' +-- and 'withKafkaProducer') addBrokers :: Kafka -> String -> IO () addBrokers (Kafka kptr _) brokerStr = do numBrokers <- rdKafkaBrokersAdd kptr brokerStr when (numBrokers == 0) (throw $ KafkaBadSpecification "No valid brokers specified") +-- | Starts consuming for a given topic. You probably do not need +-- to call this directly (it is called automatically by 'withKafkaConsumer') but +-- '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 @@ -216,12 +87,16 @@ KafkaOffset i -> i in throwOnError $ rdKafkaConsumeStart topicPtr partition trueOffset +-- | Stops consuming for a given topic. You probably do not need to call +-- this directly (it is called automatically when 'withKafkaConsumer' terminates). +stopConsuming :: KafkaTopic -> Int -> IO () +stopConsuming (KafkaTopic topicPtr _ _) partition = + throwOnError $ rdKafkaConsumeStop topicPtr partition word8PtrToBS :: Int -> Word8Ptr -> IO (BS.ByteString) word8PtrToBS len ptr = BSI.create len $ \bsptr -> BSI.memcpy bsptr ptr len - fromMessageStorable :: RdKafkaMessageT -> IO (KafkaMessage) fromMessageStorable s = do payload <- word8PtrToBS (len'RdKafkaMessageT s) (payload'RdKafkaMessageT s) @@ -234,8 +109,11 @@ (offset'RdKafkaMessageT s) payload key - -consumeMessage :: KafkaTopic -> Int -> Int -> IO (Either KafkaError KafkaMessage) +-- | Consumes a single message from a Kafka topic, waiting up to a given timeout +consumeMessage :: KafkaTopic + -> Int -- ^ partition number to consume from (must match 'withKafkaConsumer') + -> Int -- ^ the timeout, in milliseconds (10^3 per second) + -> IO (Either KafkaError KafkaMessage) -- ^ Left on error or timeout, right for success consumeMessage (KafkaTopic topicPtr _ _) partition timeout = do ptr <- rdKafkaConsume topicPtr (fromIntegral partition) (fromIntegral timeout) withForeignPtr ptr $ \realPtr -> @@ -256,7 +134,13 @@ payload key -consumeMessageBatch :: KafkaTopic -> Int -> Int -> Int -> IO (Either KafkaError [KafkaMessage]) +-- | Consumes a batch of messages from a Kafka topic, waiting up to a given timeout. Partial results +-- will be returned if a timeout occurs. +consumeMessageBatch :: KafkaTopic + -> Int -- ^ partition number to consume from (must match 'withKafkaConsumer') + -> Int -- ^ timeout in milliseconds (10^3 per second) + -> Int -- ^ maximum number of messages to return + -> IO (Either KafkaError [KafkaMessage]) -- ^ Left on error, right with up to 'maxMessages' messages on success consumeMessageBatch (KafkaTopic topicPtr _ _) partition timeout maxMessages = allocaArray maxMessages $ \outputPtr -> do numMessages <- rdKafkaConsumeBatch topicPtr (fromIntegral partition) timeout outputPtr (fromIntegral maxMessages) @@ -271,6 +155,9 @@ return ret return $ Right ms +-- | 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. storeOffset :: KafkaTopic -> Int -> Int -> IO (Maybe KafkaError) storeOffset (KafkaTopic topicPtr _ _) partition offset = do err <- rdKafkaOffsetStore topicPtr (fromIntegral partition) (fromIntegral offset) @@ -278,7 +165,13 @@ RdKafkaRespErrNoError -> return Nothing e -> return $ Just $ KafkaResponseError e -produceMessage :: KafkaTopic -> KafkaProducePartition -> KafkaProduceMessage -> IO (Maybe KafkaError) +-- | Produce a single unkeyed message to either a random partition or specified partition. Since +-- librdkafka is backed by a queue, this function can return before messages are sent. See +-- 'drainOutQueue' to wait for queue to empty. +produceMessage :: KafkaTopic -- ^ topic pointer + -> KafkaProducePartition -- ^ the partition to produce to. Specify 'KafkaUnassignedPartition' if you don't care. + -> KafkaProduceMessage -- ^ the message to enqueue. This function is undefined for keyed messages. + -> IO (Maybe KafkaError) -- Nothing on success, error if something went wrong. produceMessage (KafkaTopic topicPtr _ _) partition (KafkaProduceMessage payload) = do let (payloadFPtr, payloadOffset, payloadLength) = BSI.toForeignPtr payload @@ -292,7 +185,11 @@ produceMessage _ _ (KafkaProduceKeyedMessage _ _) = undefined -produceKeyedMessage :: KafkaTopic -> KafkaProduceMessage -> IO (Maybe KafkaError) +-- | Produce a single keyed message. Since librdkafka is backed by a queue, this function can return +-- before messages are sent. See 'drainOutQueue' to wait for a queue to be empty +produceKeyedMessage :: KafkaTopic -- ^ topic pointer + -> KafkaProduceMessage -- ^ keyed message. This function is undefined for unkeyed messages. + -> IO (Maybe KafkaError) -- ^ Nothing on success, error if something went wrong. produceKeyedMessage _ (KafkaProduceMessage _) = undefined produceKeyedMessage (KafkaTopic topicPtr _ _) (KafkaProduceKeyedMessage key payload) = do let (payloadFPtr, payloadOffset, payloadLength) = BSI.toForeignPtr payload @@ -308,7 +205,12 @@ copyMsgFlags passedPayload (fromIntegral payloadLength) passedKey (fromIntegral keyLength) nullPtr -produceMessageBatch :: KafkaTopic -> KafkaProducePartition -> [KafkaProduceMessage] -> IO ([(KafkaProduceMessage, KafkaError)]) +-- | Produce a batch of messages. Since librdkafka is backed by a queue, this function can return +-- before messages are sent. See 'drainOutQueue' to wait for the queue to be empty. +produceMessageBatch :: KafkaTopic -- ^ topic pointer + -> KafkaProducePartition -- ^ partition to produce to. Specify 'KafkaUnassignedPartition' if you don't care, or you have keyed messsages. + -> [KafkaProduceMessage] -- ^ list of messages to enqueue. + -> IO ([(KafkaProduceMessage, KafkaError)]) -- list of failed messages with their errors. This will be empty on success. produceMessageBatch (KafkaTopic topicPtr _ _) partition pms = do storables <- forM pms produceMessageToMessage withArray storables $ \batchPtr -> do @@ -353,40 +255,46 @@ , key'RdKafkaMessageT = passedKey } -type ConfigOverrides = [(String, String)] -withKafkaProducer :: ConfigOverrides -> ConfigOverrides - -> String -> String - -> (Kafka -> KafkaTopic -> IO a) - -> IO a +-- | Connects to Kafka broker in producer mode for a given topic, taking a function +-- that is fed with 'Kafka' and 'KafkaTopic' instances. After receiving handles you +-- should be using 'produceMessage', 'produceKeyedMessage' and 'produceMessageBatch' +-- to publish messages. This function drains the outbound queue automatically before returning. +withKafkaProducer :: ConfigOverrides -- ^ config overrides for kafka. See . Use an empty list if you don't care. + -> ConfigOverrides -- ^ config overrides for topic. See . Use an empty list if you don't care. + -> String -- ^ broker string, e.g. localhost:9092 + -> String -- ^ topic name + -> (Kafka -> KafkaTopic -> IO a) -- ^ your code, fed with 'Kafka' and 'KafkaTopic' instances for subsequent interaction. + -> IO a -- ^ returns what your code does withKafkaProducer configOverrides topicConfigOverrides brokerString tName cb = bracket (do - conf <- newKafkaConf - mapM_ (\(k, v) -> setKafkaConfValue conf k v) configOverrides - kafka <- newKafka KafkaProducer conf + kafka <- newKafka RdKafkaProducer configOverrides addBrokers kafka brokerString - topicConf <- newKafkaTopicConf - mapM_ (\(k, v) -> setKafkaTopicConfValue topicConf k v) topicConfigOverrides - topic <- newKafkaTopic kafka tName topicConf + topic <- newKafkaTopic kafka tName topicConfigOverrides return (kafka, topic) ) (\(kafka, _) -> drainOutQueue kafka) (\(k, t) -> cb k t) -withKafkaConsumer :: ConfigOverrides -> ConfigOverrides - -> String -> String -> Int -> KafkaOffset - -> (Kafka -> KafkaTopic -> IO a) - -> IO a +-- | Connects to Kafka broker in consumer mode for a specific partition, +-- taking a function that is fed with +-- 'Kafka' and 'KafkaTopic' instances. After receiving handles, you should be using +-- 'consumeMessage' and 'consumeMessageBatch' to receive messages. This function +-- automatically starts consuming before calling your code. +withKafkaConsumer :: ConfigOverrides -- ^ config overrides for kafka. See . Use an empty list if you don't care. + -> ConfigOverrides -- ^ config overrides for topic. See . Use an empty list if you don't care. + -> String -- ^ broker string, e.g. localhost:9092 + -> String -- ^ topic name + -> Int -- ^ partition to consume from. Locked until the function returns. + -> KafkaOffset -- ^ where to begin consuming in the partition. + -> (Kafka -> KafkaTopic -> IO a) -- ^ your cod, fed with 'Kafka' and 'KafkaTopic' instances for subsequent interaction. + -> IO a -- ^ returns what your code does withKafkaConsumer configOverrides topicConfigOverrides brokerString tName partition offset cb = bracket (do - conf <- newKafkaConf - mapM_ (\(k, v) -> setKafkaConfValue conf k v) configOverrides - kafka <- newKafka KafkaConsumer conf + kafka <- newKafka RdKafkaConsumer configOverrides addBrokers kafka brokerString - topicConf <- newKafkaTopicConf - mapM_ (\(k, v) -> setKafkaTopicConfValue topicConf k v) topicConfigOverrides - topic <- newKafkaTopic kafka tName topicConf + topic <- newKafkaTopic kafka tName topicConfigOverrides startConsuming topic partition offset return (kafka, topic) ) @@ -408,35 +316,27 @@ handleProduceErr 0 = return $ Nothing handleProduceErr _ = return $ Just $ KafkaInvalidReturnValue - -data KafkaMetadata = KafkaMetadata - { brokers :: [KafkaBrokerMetadata] - , topics :: [Either KafkaError KafkaTopicMetadata] - } deriving (Eq, Show, Typeable) - -data KafkaBrokerMetadata = KafkaBrokerMetadata - { brokerId :: Int - , brokerHost :: String - , brokerPort :: Int - } deriving (Eq, Show, Typeable) - -data KafkaTopicMetadata = KafkaTopicMetadata - { topicName :: String - , topicPartitions :: [Either KafkaError KafkaPartitionMetadata] - } deriving (Eq, Show, Typeable) - -data KafkaPartitionMetadata = KafkaPartitionMetadata - { partitionId :: Int - , partitionLeader :: Int - , partitionReplicas :: [Int] - , partitionIsrs :: [Int] - } deriving (Eq, Show, Typeable) - +-- | Opens a connection with brokers and returns metadata about topics, partitions and brokers. +fetchBrokerMetadata :: ConfigOverrides -- ^ connection overrides, see + -> String -- broker connection string, e.g. localhost:9092 + -> Int -- timeout for the request, in milliseconds (10^3 per second) + -> IO (Either KafkaError KafkaMetadata) -- Left on error, Right with metadata on success +fetchBrokerMetadata configOverrides brokerString timeout = do + kafka <- newKafka RdKafkaConsumer configOverrides + addBrokers kafka brokerString + getAllMetadata kafka timeout -getAllMetadata :: Kafka -> Int -> IO (Either KafkaError KafkaMetadata) +-- | Grabs all metadata from a given Kafka instance. +getAllMetadata :: Kafka + -> Int -- ^ timeout in milliseconds (10^3 per second) + -> IO (Either KafkaError KafkaMetadata) getAllMetadata k timeout = getMetadata k Nothing timeout -getTopicMetadata :: Kafka -> KafkaTopic -> Int -> IO (Either KafkaError KafkaTopicMetadata) +-- | Grabs topic metadata from a given Kafka topic instance +getTopicMetadata :: Kafka + -> KafkaTopic + -> Int -- ^ timeout in milliseconds (10^3 per second) + -> IO (Either KafkaError KafkaTopicMetadata) getTopicMetadata k kt timeout = do err <- getMetadata k (Just kt) timeout case err of @@ -515,6 +415,8 @@ outboundQueueLength :: Kafka -> IO (Int) outboundQueueLength (Kafka kPtr _) = rdKafkaOutqLen kPtr +-- | Drains the outbound queue for a producer. This function is called automatically at the end of +-- 'withKafkaProducer' and usually doesn't need to be called directly. drainOutQueue :: Kafka -> IO () drainOutQueue k = do pollEvents k 100 @@ -525,10 +427,6 @@ kafkaRespErr :: Errno -> KafkaError kafkaRespErr (Errno num) = KafkaResponseError $ rdKafkaErrno2err (fromIntegral num) -stopConsuming :: KafkaTopic -> Int -> IO () -stopConsuming (KafkaTopic topicPtr _ _) partition = - throwOnError $ rdKafkaConsumeStop topicPtr partition - throwOnError :: IO (Maybe String) -> IO () throwOnError action = do m <- action @@ -536,34 +434,3 @@ Just e -> throw $ KafkaError e Nothing -> return () - -dumpKafkaTopicConf :: KafkaTopicConf -> IO (Map String String) -dumpKafkaTopicConf (KafkaTopicConf kptr) = - parseDump (\sizeptr -> rdKafkaTopicConfDump kptr sizeptr) - -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 - strPtr <- cstr sizeptr - size <- peek sizeptr - - keysAndValues <- mapM (\i -> peekCString =<< peekElemOff strPtr i) [0..((fromIntegral size) - 1)] - - let ret = Map.fromList $ listToTuple keysAndValues - rdKafkaConfDumpFree strPtr size - return ret - -listToTuple :: [String] -> [(String, String)] -listToTuple [] = [] -listToTuple (k:v:t) = (k, v) : listToTuple t -listToTuple _ = error "list to tuple can only be called on even length lists" diff --git a/src/Haskakafka/Example.hs b/src/Haskakafka/Example.hs --- a/src/Haskakafka/Example.hs +++ b/src/Haskakafka/Example.hs @@ -61,4 +61,11 @@ -- Be a little less noisy - setKafkaLogLevel kafka KafkaLogCrit + setLogLevel kafka KafkaLogCrit + + -- we can also fetch metadata about our Kafka infrastructure + let timeoutMs = 1000 + emd <- fetchBrokerMetadata [] "localhost:9092" timeoutMs + case emd of + (Left err) -> putStrLn $ "Uh oh, error time: " ++ (show err) + (Right md) -> putStrLn $ "Kafka metadata: " ++ (show md) diff --git a/src/Haskakafka/Internal.chs b/src/Haskakafka/Internal.chs deleted file mode 100644 --- a/src/Haskakafka/Internal.chs +++ /dev/null @@ -1,377 +0,0 @@ -{-# LANGUAGE ForeignFunctionInterface #-} -{-# LANGUAGE EmptyDataDecls #-} - -module Haskakafka.Internal where - -import Control.Applicative -import Control.Monad -import Data.Word -import Foreign hiding (unsafePerformIO) -import Foreign.C.Error -import Foreign.C.String -import Foreign.C.Types -import Haskakafka.InternalEnum -import System.IO -import System.IO.Unsafe (unsafePerformIO) -import System.Posix.IO -import System.Posix.Types - -#include "rdkafka.h" - -type CInt64T = {#type int64_t #} -type CInt32T = {#type int32_t #} - -{#pointer *FILE as CFilePtr -> CFile #} -{#pointer *size_t as CSizePtr -> CSize #} - -type Word8Ptr = Ptr Word8 -type CCharBufPointer = Ptr CChar - -type RdKafkaMsgFlag = Int -rdKafkaMsgFlagFree :: RdKafkaMsgFlag -rdKafkaMsgFlagFree = 0x1 -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' #} - -{#fun pure unsafe rd_kafka_version_str as ^ - {} -> `String' #} - -{#fun pure unsafe rd_kafka_err2str as ^ - {enumToCInt `RdKafkaRespErrT'} -> `String' #} - -{#fun pure unsafe rd_kafka_errno2err as ^ - {`Int'} -> `RdKafkaRespErrT' cIntToEnum #} - - -kafkaErrnoString :: IO (String) -kafkaErrnoString = do - (Errno num) <- getErrno - return $ rdKafkaErr2str $ rdKafkaErrno2err (fromIntegral num) - --- Kafka Pointer Types -data RdKafkaConfT -{#pointer *rd_kafka_conf_t as RdKafkaConfTPtr foreign -> RdKafkaConfT #} - -data RdKafkaTopicConfT -{#pointer *rd_kafka_topic_conf_t as RdKafkaTopicConfTPtr foreign -> RdKafkaTopicConfT #} - -data RdKafkaT -{#pointer *rd_kafka_t as RdKafkaTPtr foreign -> RdKafkaT #} - -data RdKafkaTopicT -{#pointer *rd_kafka_topic_t as RdKafkaTopicTPtr foreign -> RdKafkaTopicT #} - -data RdKafkaMessageT = RdKafkaMessageT - { err'RdKafkaMessageT :: RdKafkaRespErrT - , partition'RdKafkaMessageT :: Int - , len'RdKafkaMessageT :: Int - , keyLen'RdKafkaMessageT :: Int - , offset'RdKafkaMessageT :: Int64 - , payload'RdKafkaMessageT :: Word8Ptr - , key'RdKafkaMessageT :: Word8Ptr - } - deriving (Show, Eq) - -instance Storable RdKafkaMessageT where - alignment _ = {#alignof rd_kafka_message_t#} - sizeOf _ = {#sizeof rd_kafka_message_t#} - peek p = RdKafkaMessageT - <$> liftM cIntToEnum ({#get rd_kafka_message_t->err #} p) - <*> liftM fromIntegral ({#get rd_kafka_message_t->partition #} p) - <*> liftM fromIntegral ({#get rd_kafka_message_t->len #} p) - <*> liftM fromIntegral ({#get rd_kafka_message_t->key_len#} p) - <*> 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) - poke p x = do - {#set rd_kafka_message_t.err#} p (enumToCInt $ err'RdKafkaMessageT x) - {#set rd_kafka_message_t.partition#} p (fromIntegral $ partition'RdKafkaMessageT x) - {#set rd_kafka_message_t.len#} p (fromIntegral $ len'RdKafkaMessageT x) - {#set rd_kafka_message_t.key_len#} p (fromIntegral $ keyLen'RdKafkaMessageT x) - {#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) - -{#pointer *rd_kafka_message_t as RdKafkaMessageTPtr foreign -> RdKafkaMessageT #} - -data RdKafkaMetadataBrokerT = RdKafkaMetadataBrokerT - { id'RdKafkaMetadataBrokerT :: Int - , host'RdKafkaMetadataBrokerT :: CString - , port'RdKafkaMetadataBrokerT :: Int - } deriving (Show, Eq) - -{#pointer *rd_kafka_metadata_broker_t as RdKafkaMetadataBrokerTPtr -> RdKafkaMetadataBrokerT #} - - -instance Storable RdKafkaMetadataBrokerT where - alignment _ = {#alignof rd_kafka_metadata_broker_t#} - 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) - <*> liftM fromIntegral ({#get rd_kafka_metadata_broker_t->port #} p) - poke = undefined - -data RdKafkaMetadataPartitionT = RdKafkaMetadataPartitionT - { id'RdKafkaMetadataPartitionT :: Int - , err'RdKafkaMetadataPartitionT :: RdKafkaRespErrT - , leader'RdKafkaMetadataPartitionT :: Int - , replicaCnt'RdKafkaMetadataPartitionT :: Int - , replicas'RdKafkaMetadataPartitionT :: Ptr CInt32T - , isrCnt'RdKafkaMetadataPartitionT :: Int - , isrs'RdKafkaMetadataPartitionT :: Ptr CInt32T - } deriving (Show, Eq) - -instance Storable RdKafkaMetadataPartitionT where - alignment _ = {#alignof rd_kafka_metadata_partition_t#} - sizeOf _ = {#sizeof rd_kafka_metadata_partition_t#} - peek p = RdKafkaMetadataPartitionT - <$> liftM fromIntegral ({#get rd_kafka_metadata_partition_t->id#} p) - <*> liftM cIntToEnum ({#get rd_kafka_metadata_partition_t->err#} p) - <*> liftM fromIntegral ({#get rd_kafka_metadata_partition_t->leader#} p) - <*> liftM fromIntegral ({#get rd_kafka_metadata_partition_t->replica_cnt#} p) - <*> liftM castPtr ({#get rd_kafka_metadata_partition_t->replicas#} p) - <*> liftM fromIntegral ({#get rd_kafka_metadata_partition_t->isr_cnt#} p) - <*> liftM castPtr ({#get rd_kafka_metadata_partition_t->isrs#} p) - - poke = undefined - -{#pointer *rd_kafka_metadata_partition_t as RdKafkaMetadataPartitionTPtr -> RdKafkaMetadataPartitionT #} - -data RdKafkaMetadataTopicT = RdKafkaMetadataTopicT - { topic'RdKafkaMetadataTopicT :: CString - , partitionCnt'RdKafkaMetadataTopicT :: Int - , partitions'RdKafkaMetadataTopicT :: Ptr RdKafkaMetadataPartitionT - , err'RdKafkaMetadataTopicT :: RdKafkaRespErrT - } deriving (Show, Eq) - -instance Storable RdKafkaMetadataTopicT where - alignment _ = {#alignof rd_kafka_metadata_topic_t#} - 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) - <*> liftM castPtr ({#get rd_kafka_metadata_topic_t->partitions #} p) - <*> liftM cIntToEnum ({#get rd_kafka_metadata_topic_t->err #} p) - poke p x = undefined - -{#pointer *rd_kafka_metadata_topic_t as RdKafkaMetadataTopicTPtr -> RdKafkaMetadataTopicT #} - -data RdKafkaMetadataT = RdKafkaMetadataT - { brokerCnt'RdKafkaMetadataT :: Int - , brokers'RdKafkaMetadataT :: RdKafkaMetadataBrokerTPtr - , topicCnt'RdKafkaMetadataT :: Int - , topics'RdKafkaMetadataT :: RdKafkaMetadataTopicTPtr - , origBrokerId'RdKafkaMetadataT :: CInt32T - } deriving (Show, Eq) - -instance Storable RdKafkaMetadataT where - alignment _ = {#alignof rd_kafka_metadata_t#} - sizeOf _ = {#sizeof rd_kafka_metadata_t#} - peek p = RdKafkaMetadataT - <$> liftM fromIntegral ({#get rd_kafka_metadata_t->broker_cnt #} p) - <*> liftM castPtr ({#get rd_kafka_metadata_t->brokers #} p) - <*> liftM fromIntegral ({#get rd_kafka_metadata_t->topic_cnt #} p) - <*> liftM castPtr ({#get rd_kafka_metadata_t->topics #} p) - <*> liftM fromIntegral ({#get rd_kafka_metadata_t->orig_broker_id #} p) - poke p x = undefined - -{#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' #} - -foreign import ccall unsafe "rdkafka.h &rd_kafka_conf_destroy" - rdKafkaConfDestroy :: FunPtr (Ptr RdKafkaConfT -> IO ()) - -{#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 - addForeignPtrFinalizer rdKafkaConfDestroy ret - return ret - -{#fun unsafe rd_kafka_conf_dump as ^ - {`RdKafkaConfTPtr', castPtr `CSizePtr'} -> `Ptr CString' id #} - -{#fun unsafe rd_kafka_conf_dump_free as ^ - {id `Ptr CString', cIntConv `CSize'} -> `()' #} - -{#fun unsafe rd_kafka_conf_properties_show as ^ - {`CFilePtr'} -> `()' #} - --- rd_kafka_topic_conf -{#fun unsafe rd_kafka_topic_conf_new as ^ - {} -> `RdKafkaTopicConfTPtr' #} -{#fun unsafe rd_kafka_topic_conf_dup as ^ - {`RdKafkaTopicConfTPtr'} -> `RdKafkaTopicConfTPtr' #} - -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 - addForeignPtrFinalizer rdKafkaTopicConfDestroy ret - return ret - -{#fun unsafe rd_kafka_topic_conf_dump as ^ - {`RdKafkaTopicConfTPtr', castPtr `CSizePtr'} -> `Ptr CString' id #} - --- rd_kafka -{#fun unsafe rd_kafka_new as ^ - {enumToCInt `RdKafkaTypeT', `RdKafkaConfTPtr', id `CCharBufPointer', cIntConv `CSize'} - -> `RdKafkaTPtr' #} - -foreign import ccall unsafe "rdkafka.h &rd_kafka_destroy" - rdKafkaDestroy :: FunPtr (Ptr RdKafkaT -> IO ()) - -newRdKafkaT :: RdKafkaTypeT -> RdKafkaConfTPtr -> IO (Either String RdKafkaTPtr) -newRdKafkaT kafkaType confPtr = - allocaBytes nErrorBytes $ \charPtr -> do - duper <- rdKafkaConfDup confPtr - ret <- rdKafkaNew kafkaType duper charPtr (fromIntegral nErrorBytes) - withForeignPtr ret $ \realPtr -> do - if realPtr == nullPtr then peekCString charPtr >>= return . Left - else do - addForeignPtrFinalizer rdKafkaDestroy ret - return $ Right ret - -{#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' #} - -rdKafkaConsumeStart :: RdKafkaTopicTPtr -> Int -> Int64 -> IO (Maybe String) -rdKafkaConsumeStart topicPtr partition offset = do - i <- rdKafkaConsumeStartInternal topicPtr (fromIntegral partition) (fromIntegral offset) - case i of - -1 -> kafkaErrnoString >>= return . Just - _ -> return Nothing -{#fun unsafe rd_kafka_consume_stop as rdKafkaConsumeStopInternal - {`RdKafkaTopicTPtr', cIntConv `CInt32T'} -> `Int' #} - -{#fun unsafe rd_kafka_consume as ^ - {`RdKafkaTopicTPtr', cIntConv `CInt32T', `Int'} -> `RdKafkaMessageTPtr' #} - -{#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 - i <- rdKafkaConsumeStopInternal topicPtr (fromIntegral partition) - case i of - -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'} - -> `Int' #} - -{#fun unsafe rd_kafka_produce_batch as ^ - {`RdKafkaTopicTPtr', cIntConv `CInt32T', `Int', `RdKafkaMessageTPtr', `Int'} -> `Int' #} - -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'} - -> `RdKafkaRespErrT' cIntToEnum #} - -{# fun unsafe rd_kafka_metadata_destroy as ^ - {castPtr `Ptr RdKafkaMetadataT'} -> `()' #} - -{#fun unsafe rd_kafka_poll as ^ - {`RdKafkaTPtr', `Int'} -> `Int' #} - -{#fun unsafe rd_kafka_outq_len as ^ - {`RdKafkaTPtr'} -> `Int' #} - -{#fun unsafe rd_kafka_dump as ^ - {`CFilePtr', `RdKafkaTPtr'} -> `()' #} - - --- rd_kafka_topic -{#fun unsafe rd_kafka_topic_new as ^ - {`RdKafkaTPtr', `String', `RdKafkaTopicConfTPtr'} -> `RdKafkaTopicTPtr' #} - -foreign import ccall unsafe "rdkafka.h &rd_kafka_topic_destroy" - rdKafkaTopicDestroy :: FunPtr (Ptr RdKafkaTopicT -> IO ()) - -newRdKafkaTopicT :: RdKafkaTPtr -> String -> RdKafkaTopicConfTPtr -> IO (Either String RdKafkaTopicTPtr) -newRdKafkaTopicT kafkaPtr topic topicConfPtr = do - duper <- rdKafkaTopicConfDup topicConfPtr - ret <- rdKafkaTopicNew kafkaPtr topic duper - withForeignPtr ret $ \realPtr -> - if realPtr == nullPtr then kafkaErrnoString >>= return . Left - else do - addForeignPtrFinalizer rdKafkaTopicDestroy ret - return $ Right ret - --- Marshall / Unmarshall -enumToCInt :: Enum a => a -> CInt -enumToCInt = fromIntegral . fromEnum -cIntToEnum :: Enum a => CInt -> a -cIntToEnum = toEnum . fromIntegral -cIntConv :: (Integral a, Num b) => a -> b -cIntConv = fromIntegral -boolToCInt :: Bool -> CInt -boolToCInt True = CInt 1 -boolToCInt False = CInt 0 - --- Handle -> File descriptor - -foreign import ccall "" fdopen :: Fd -> CString -> IO (Ptr CFile) - -handleToCFile :: Handle -> String -> IO (CFilePtr) -handleToCFile h m = - do iomode <- newCString m - fd <- handleToFd h - fdopen fd iomode - -c_stdin :: IO CFilePtr -c_stdin = handleToCFile stdin "r" -c_stdout :: IO CFilePtr -c_stdout = handleToCFile stdout "w" -c_stderr :: IO CFilePtr -c_stderr = handleToCFile stderr "w" diff --git a/src/Haskakafka/InternalEnum.chs b/src/Haskakafka/InternalEnum.chs deleted file mode 100644 --- a/src/Haskakafka/InternalEnum.chs +++ /dev/null @@ -1,10 +0,0 @@ -{-# LANGUAGE ForeignFunctionInterface #-} -{-# LANGUAGE EmptyDataDecls #-} - -module Haskakafka.InternalEnum where - -#include "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) #} diff --git a/src/Haskakafka/InternalRdKafka.chs b/src/Haskakafka/InternalRdKafka.chs new file mode 100644 --- /dev/null +++ b/src/Haskakafka/InternalRdKafka.chs @@ -0,0 +1,377 @@ +{-# LANGUAGE ForeignFunctionInterface #-} +{-# LANGUAGE EmptyDataDecls #-} + +module Haskakafka.InternalRdKafka where + +import Control.Applicative +import Control.Monad +import Data.Word +import Foreign hiding (unsafePerformIO) +import Foreign.C.Error +import Foreign.C.String +import Foreign.C.Types +import Haskakafka.InternalRdKafkaEnum +import System.IO +import System.IO.Unsafe (unsafePerformIO) +import System.Posix.IO +import System.Posix.Types + +#include "rdkafka.h" + +type CInt64T = {#type int64_t #} +type CInt32T = {#type int32_t #} + +{#pointer *FILE as CFilePtr -> CFile #} +{#pointer *size_t as CSizePtr -> CSize #} + +type Word8Ptr = Ptr Word8 +type CCharBufPointer = Ptr CChar + +type RdKafkaMsgFlag = Int +rdKafkaMsgFlagFree :: RdKafkaMsgFlag +rdKafkaMsgFlagFree = 0x1 +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' #} + +{#fun pure unsafe rd_kafka_version_str as ^ + {} -> `String' #} + +{#fun pure unsafe rd_kafka_err2str as ^ + {enumToCInt `RdKafkaRespErrT'} -> `String' #} + +{#fun pure unsafe rd_kafka_errno2err as ^ + {`Int'} -> `RdKafkaRespErrT' cIntToEnum #} + + +kafkaErrnoString :: IO (String) +kafkaErrnoString = do + (Errno num) <- getErrno + return $ rdKafkaErr2str $ rdKafkaErrno2err (fromIntegral num) + +-- Kafka Pointer Types +data RdKafkaConfT +{#pointer *rd_kafka_conf_t as RdKafkaConfTPtr foreign -> RdKafkaConfT #} + +data RdKafkaTopicConfT +{#pointer *rd_kafka_topic_conf_t as RdKafkaTopicConfTPtr foreign -> RdKafkaTopicConfT #} + +data RdKafkaT +{#pointer *rd_kafka_t as RdKafkaTPtr foreign -> RdKafkaT #} + +data RdKafkaTopicT +{#pointer *rd_kafka_topic_t as RdKafkaTopicTPtr foreign -> RdKafkaTopicT #} + +data RdKafkaMessageT = RdKafkaMessageT + { err'RdKafkaMessageT :: RdKafkaRespErrT + , partition'RdKafkaMessageT :: Int + , len'RdKafkaMessageT :: Int + , keyLen'RdKafkaMessageT :: Int + , offset'RdKafkaMessageT :: Int64 + , payload'RdKafkaMessageT :: Word8Ptr + , key'RdKafkaMessageT :: Word8Ptr + } + deriving (Show, Eq) + +instance Storable RdKafkaMessageT where + alignment _ = {#alignof rd_kafka_message_t#} + sizeOf _ = {#sizeof rd_kafka_message_t#} + peek p = RdKafkaMessageT + <$> liftM cIntToEnum ({#get rd_kafka_message_t->err #} p) + <*> liftM fromIntegral ({#get rd_kafka_message_t->partition #} p) + <*> liftM fromIntegral ({#get rd_kafka_message_t->len #} p) + <*> liftM fromIntegral ({#get rd_kafka_message_t->key_len#} p) + <*> 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) + poke p x = do + {#set rd_kafka_message_t.err#} p (enumToCInt $ err'RdKafkaMessageT x) + {#set rd_kafka_message_t.partition#} p (fromIntegral $ partition'RdKafkaMessageT x) + {#set rd_kafka_message_t.len#} p (fromIntegral $ len'RdKafkaMessageT x) + {#set rd_kafka_message_t.key_len#} p (fromIntegral $ keyLen'RdKafkaMessageT x) + {#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) + +{#pointer *rd_kafka_message_t as RdKafkaMessageTPtr foreign -> RdKafkaMessageT #} + +data RdKafkaMetadataBrokerT = RdKafkaMetadataBrokerT + { id'RdKafkaMetadataBrokerT :: Int + , host'RdKafkaMetadataBrokerT :: CString + , port'RdKafkaMetadataBrokerT :: Int + } deriving (Show, Eq) + +{#pointer *rd_kafka_metadata_broker_t as RdKafkaMetadataBrokerTPtr -> RdKafkaMetadataBrokerT #} + + +instance Storable RdKafkaMetadataBrokerT where + alignment _ = {#alignof rd_kafka_metadata_broker_t#} + 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) + <*> liftM fromIntegral ({#get rd_kafka_metadata_broker_t->port #} p) + poke = undefined + +data RdKafkaMetadataPartitionT = RdKafkaMetadataPartitionT + { id'RdKafkaMetadataPartitionT :: Int + , err'RdKafkaMetadataPartitionT :: RdKafkaRespErrT + , leader'RdKafkaMetadataPartitionT :: Int + , replicaCnt'RdKafkaMetadataPartitionT :: Int + , replicas'RdKafkaMetadataPartitionT :: Ptr CInt32T + , isrCnt'RdKafkaMetadataPartitionT :: Int + , isrs'RdKafkaMetadataPartitionT :: Ptr CInt32T + } deriving (Show, Eq) + +instance Storable RdKafkaMetadataPartitionT where + alignment _ = {#alignof rd_kafka_metadata_partition_t#} + sizeOf _ = {#sizeof rd_kafka_metadata_partition_t#} + peek p = RdKafkaMetadataPartitionT + <$> liftM fromIntegral ({#get rd_kafka_metadata_partition_t->id#} p) + <*> liftM cIntToEnum ({#get rd_kafka_metadata_partition_t->err#} p) + <*> liftM fromIntegral ({#get rd_kafka_metadata_partition_t->leader#} p) + <*> liftM fromIntegral ({#get rd_kafka_metadata_partition_t->replica_cnt#} p) + <*> liftM castPtr ({#get rd_kafka_metadata_partition_t->replicas#} p) + <*> liftM fromIntegral ({#get rd_kafka_metadata_partition_t->isr_cnt#} p) + <*> liftM castPtr ({#get rd_kafka_metadata_partition_t->isrs#} p) + + poke = undefined + +{#pointer *rd_kafka_metadata_partition_t as RdKafkaMetadataPartitionTPtr -> RdKafkaMetadataPartitionT #} + +data RdKafkaMetadataTopicT = RdKafkaMetadataTopicT + { topic'RdKafkaMetadataTopicT :: CString + , partitionCnt'RdKafkaMetadataTopicT :: Int + , partitions'RdKafkaMetadataTopicT :: Ptr RdKafkaMetadataPartitionT + , err'RdKafkaMetadataTopicT :: RdKafkaRespErrT + } deriving (Show, Eq) + +instance Storable RdKafkaMetadataTopicT where + alignment _ = {#alignof rd_kafka_metadata_topic_t#} + 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) + <*> liftM castPtr ({#get rd_kafka_metadata_topic_t->partitions #} p) + <*> liftM cIntToEnum ({#get rd_kafka_metadata_topic_t->err #} p) + poke p x = undefined + +{#pointer *rd_kafka_metadata_topic_t as RdKafkaMetadataTopicTPtr -> RdKafkaMetadataTopicT #} + +data RdKafkaMetadataT = RdKafkaMetadataT + { brokerCnt'RdKafkaMetadataT :: Int + , brokers'RdKafkaMetadataT :: RdKafkaMetadataBrokerTPtr + , topicCnt'RdKafkaMetadataT :: Int + , topics'RdKafkaMetadataT :: RdKafkaMetadataTopicTPtr + , origBrokerId'RdKafkaMetadataT :: CInt32T + } deriving (Show, Eq) + +instance Storable RdKafkaMetadataT where + alignment _ = {#alignof rd_kafka_metadata_t#} + sizeOf _ = {#sizeof rd_kafka_metadata_t#} + peek p = RdKafkaMetadataT + <$> liftM fromIntegral ({#get rd_kafka_metadata_t->broker_cnt #} p) + <*> liftM castPtr ({#get rd_kafka_metadata_t->brokers #} p) + <*> liftM fromIntegral ({#get rd_kafka_metadata_t->topic_cnt #} p) + <*> liftM castPtr ({#get rd_kafka_metadata_t->topics #} p) + <*> liftM fromIntegral ({#get rd_kafka_metadata_t->orig_broker_id #} p) + poke p x = undefined + +{#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' #} + +foreign import ccall unsafe "rdkafka.h &rd_kafka_conf_destroy" + rdKafkaConfDestroy :: FunPtr (Ptr RdKafkaConfT -> IO ()) + +{#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 + addForeignPtrFinalizer rdKafkaConfDestroy ret + return ret + +{#fun unsafe rd_kafka_conf_dump as ^ + {`RdKafkaConfTPtr', castPtr `CSizePtr'} -> `Ptr CString' id #} + +{#fun unsafe rd_kafka_conf_dump_free as ^ + {id `Ptr CString', cIntConv `CSize'} -> `()' #} + +{#fun unsafe rd_kafka_conf_properties_show as ^ + {`CFilePtr'} -> `()' #} + +-- rd_kafka_topic_conf +{#fun unsafe rd_kafka_topic_conf_new as ^ + {} -> `RdKafkaTopicConfTPtr' #} +{#fun unsafe rd_kafka_topic_conf_dup as ^ + {`RdKafkaTopicConfTPtr'} -> `RdKafkaTopicConfTPtr' #} + +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 + addForeignPtrFinalizer rdKafkaTopicConfDestroy ret + return ret + +{#fun unsafe rd_kafka_topic_conf_dump as ^ + {`RdKafkaTopicConfTPtr', castPtr `CSizePtr'} -> `Ptr CString' id #} + +-- rd_kafka +{#fun unsafe rd_kafka_new as ^ + {enumToCInt `RdKafkaTypeT', `RdKafkaConfTPtr', id `CCharBufPointer', cIntConv `CSize'} + -> `RdKafkaTPtr' #} + +foreign import ccall unsafe "rdkafka.h &rd_kafka_destroy" + rdKafkaDestroy :: FunPtr (Ptr RdKafkaT -> IO ()) + +newRdKafkaT :: RdKafkaTypeT -> RdKafkaConfTPtr -> IO (Either String RdKafkaTPtr) +newRdKafkaT kafkaType confPtr = + allocaBytes nErrorBytes $ \charPtr -> do + duper <- rdKafkaConfDup confPtr + ret <- rdKafkaNew kafkaType duper charPtr (fromIntegral nErrorBytes) + withForeignPtr ret $ \realPtr -> do + if realPtr == nullPtr then peekCString charPtr >>= return . Left + else do + addForeignPtrFinalizer rdKafkaDestroy ret + return $ Right ret + +{#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' #} + +rdKafkaConsumeStart :: RdKafkaTopicTPtr -> Int -> Int64 -> IO (Maybe String) +rdKafkaConsumeStart topicPtr partition offset = do + i <- rdKafkaConsumeStartInternal topicPtr (fromIntegral partition) (fromIntegral offset) + case i of + -1 -> kafkaErrnoString >>= return . Just + _ -> return Nothing +{#fun unsafe rd_kafka_consume_stop as rdKafkaConsumeStopInternal + {`RdKafkaTopicTPtr', cIntConv `CInt32T'} -> `Int' #} + +{#fun unsafe rd_kafka_consume as ^ + {`RdKafkaTopicTPtr', cIntConv `CInt32T', `Int'} -> `RdKafkaMessageTPtr' #} + +{#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 + i <- rdKafkaConsumeStopInternal topicPtr (fromIntegral partition) + case i of + -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'} + -> `Int' #} + +{#fun unsafe rd_kafka_produce_batch as ^ + {`RdKafkaTopicTPtr', cIntConv `CInt32T', `Int', `RdKafkaMessageTPtr', `Int'} -> `Int' #} + +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'} + -> `RdKafkaRespErrT' cIntToEnum #} + +{# fun unsafe rd_kafka_metadata_destroy as ^ + {castPtr `Ptr RdKafkaMetadataT'} -> `()' #} + +{#fun unsafe rd_kafka_poll as ^ + {`RdKafkaTPtr', `Int'} -> `Int' #} + +{#fun unsafe rd_kafka_outq_len as ^ + {`RdKafkaTPtr'} -> `Int' #} + +{#fun unsafe rd_kafka_dump as ^ + {`CFilePtr', `RdKafkaTPtr'} -> `()' #} + + +-- rd_kafka_topic +{#fun unsafe rd_kafka_topic_new as ^ + {`RdKafkaTPtr', `String', `RdKafkaTopicConfTPtr'} -> `RdKafkaTopicTPtr' #} + +foreign import ccall unsafe "rdkafka.h &rd_kafka_topic_destroy" + rdKafkaTopicDestroy :: FunPtr (Ptr RdKafkaTopicT -> IO ()) + +newRdKafkaTopicT :: RdKafkaTPtr -> String -> RdKafkaTopicConfTPtr -> IO (Either String RdKafkaTopicTPtr) +newRdKafkaTopicT kafkaPtr topic topicConfPtr = do + duper <- rdKafkaTopicConfDup topicConfPtr + ret <- rdKafkaTopicNew kafkaPtr topic duper + withForeignPtr ret $ \realPtr -> + if realPtr == nullPtr then kafkaErrnoString >>= return . Left + else do + addForeignPtrFinalizer rdKafkaTopicDestroy ret + return $ Right ret + +-- Marshall / Unmarshall +enumToCInt :: Enum a => a -> CInt +enumToCInt = fromIntegral . fromEnum +cIntToEnum :: Enum a => CInt -> a +cIntToEnum = toEnum . fromIntegral +cIntConv :: (Integral a, Num b) => a -> b +cIntConv = fromIntegral +boolToCInt :: Bool -> CInt +boolToCInt True = CInt 1 +boolToCInt False = CInt 0 + +-- Handle -> File descriptor + +foreign import ccall "" fdopen :: Fd -> CString -> IO (Ptr CFile) + +handleToCFile :: Handle -> String -> IO (CFilePtr) +handleToCFile h m = + do iomode <- newCString m + fd <- handleToFd h + fdopen fd iomode + +c_stdin :: IO CFilePtr +c_stdin = handleToCFile stdin "r" +c_stdout :: IO CFilePtr +c_stdout = handleToCFile stdout "w" +c_stderr :: IO CFilePtr +c_stderr = handleToCFile stderr "w" diff --git a/src/Haskakafka/InternalRdKafkaEnum.chs b/src/Haskakafka/InternalRdKafkaEnum.chs new file mode 100644 --- /dev/null +++ b/src/Haskakafka/InternalRdKafkaEnum.chs @@ -0,0 +1,10 @@ +{-# LANGUAGE ForeignFunctionInterface #-} +{-# LANGUAGE EmptyDataDecls #-} + +module Haskakafka.InternalRdKafkaEnum where + +#include "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) #} diff --git a/src/Haskakafka/InternalSetup.hs b/src/Haskakafka/InternalSetup.hs new file mode 100644 --- /dev/null +++ b/src/Haskakafka/InternalSetup.hs @@ -0,0 +1,145 @@ +module Haskakafka.InternalSetup where + +import Haskakafka.InternalTypes +import Haskakafka.InternalRdKafka +import Haskakafka.InternalRdKafkaEnum + +import Control.Exception +import Control.Monad +import Data.Map.Strict (Map) +import Foreign +import Foreign.C.String +import System.IO + +import qualified Data.Map.Strict as Map + +-- | Create kafka object with the given configuration. Most of the time +-- you will not need to use this function directly +-- (see 'withKafkaProducer' and 'withKafkaConsumer'). +newKafka :: RdKafkaTypeT -> ConfigOverrides -> IO Kafka +newKafka kafkaType overrides = (kafkaConf overrides) >>= newKafkaPtr kafkaType + +-- | Create a kafka topic object with the given configuration. Most of the +-- time you will not need to use this function directly +-- (see 'withKafkaProducer' and 'withKafkaConsumer') +newKafkaTopic :: Kafka -> String -> ConfigOverrides -> IO KafkaTopic +newKafkaTopic k tName overrides = (kafkaTopicConf overrides) >>= newKafkaTopicPtr k tName + +newKafkaPtr :: RdKafkaTypeT -> KafkaConf -> IO Kafka +newKafkaPtr kafkaType c@(KafkaConf confPtr) = do + et <- newRdKafkaT kafkaType confPtr + case et of + Left e -> error e + Right x -> return $ Kafka x c + +newKafkaTopicPtr :: Kafka -> String -> KafkaTopicConf -> IO (KafkaTopic) +newKafkaTopicPtr 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 conf + +-- +-- Misc. +-- +-- | Sets library log level (noisiness) with respect to a kafka instance +setLogLevel :: Kafka -> KafkaLogLevel -> IO () +setLogLevel (Kafka kptr _) level = + rdKafkaSetLogLevel kptr (fromEnum level) + +-- +-- Configuration +-- + +-- | Used to override default config properties for consumers and producers +type ConfigOverrides = [(String, String)] + +newKafkaTopicConf :: IO KafkaTopicConf +newKafkaTopicConf = newRdKafkaTopicConfT >>= return . KafkaTopicConf + +newKafkaConf :: IO KafkaConf +newKafkaConf = newRdKafkaConfT >>= return . KafkaConf + +kafkaConf :: ConfigOverrides -> IO (KafkaConf) +kafkaConf overrides = do + conf <- newKafkaConf + setAllKafkaConfValues conf overrides + return conf + +kafkaTopicConf :: ConfigOverrides -> IO (KafkaTopicConf) +kafkaTopicConf overrides = do + conf <- newKafkaTopicConf + setAllKafkaTopicConfValues conf overrides + return conf + +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 + +setAllKafkaConfValues :: KafkaConf -> ConfigOverrides -> IO () +setAllKafkaConfValues conf overrides = forM_ overrides $ \(k, v) -> setKafkaConfValue conf k v + +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 + +setAllKafkaTopicConfValues :: KafkaTopicConf -> ConfigOverrides -> IO () +setAllKafkaTopicConfValues conf overrides = forM_ overrides $ \(k, v) -> setKafkaTopicConfValue conf k v + + +-- +-- Dumping +-- +-- | Prints out all supported Kafka conf properties to a handle +hPrintSupportedKafkaConf :: Handle -> IO () +hPrintSupportedKafkaConf h = handleToCFile h "w" >>= rdKafkaConfPropertiesShow + +-- | Prints out all data associated with a specific kafka object to a handle +hPrintKafka :: Handle -> Kafka -> IO () +hPrintKafka h k = handleToCFile h "w" >>= \f -> rdKafkaDump f (kafkaPtr k) + +-- | Returns a map of the current kafka configuration +dumpConfFromKafka :: Kafka -> IO (Map String String) +dumpConfFromKafka (Kafka _ cfg) = dumpKafkaConf cfg + +-- | Returns a map of the current topic configuration +dumpConfFromKafkaTopic :: KafkaTopic -> IO (Map String String) +dumpConfFromKafkaTopic (KafkaTopic _ _ conf) = dumpKafkaTopicConf conf + +dumpKafkaTopicConf :: KafkaTopicConf -> IO (Map String String) +dumpKafkaTopicConf (KafkaTopicConf kptr) = + parseDump (\sizeptr -> rdKafkaTopicConfDump kptr sizeptr) + +dumpKafkaConf :: KafkaConf -> IO (Map String String) +dumpKafkaConf (KafkaConf kptr) = do + parseDump (\sizeptr -> rdKafkaConfDump kptr sizeptr) + +parseDump :: (CSizePtr -> IO (Ptr CString)) -> IO (Map String String) +parseDump cstr = alloca $ \sizeptr -> do + strPtr <- cstr sizeptr + size <- peek sizeptr + + keysAndValues <- mapM (\i -> peekCString =<< peekElemOff strPtr i) [0..((fromIntegral size) - 1)] + + let ret = Map.fromList $ listToTuple keysAndValues + rdKafkaConfDumpFree strPtr size + return ret + +listToTuple :: [String] -> [(String, String)] +listToTuple [] = [] +listToTuple (k:v:t) = (k, v) : listToTuple t +listToTuple _ = error "list to tuple can only be called on even length lists" diff --git a/src/Haskakafka/InternalTypes.hs b/src/Haskakafka/InternalTypes.hs new file mode 100644 --- /dev/null +++ b/src/Haskakafka/InternalTypes.hs @@ -0,0 +1,184 @@ +{-# LANGUAGE DeriveDataTypeable #-} + +module Haskakafka.InternalTypes where + +import Control.Exception +import Data.Int +import Data.Typeable + +import Haskakafka.InternalRdKafka +import Haskakafka.InternalRdKafkaEnum + +import qualified Data.ByteString as BS + +-- +-- Pointer wrappers +-- + +-- | Kafka configuration object +data KafkaConf = KafkaConf RdKafkaConfTPtr + +-- | Kafka topic configuration object +data KafkaTopicConf = KafkaTopicConf RdKafkaTopicConfTPtr + + +-- | Main pointer to Kafka object, which contains our brokers +data Kafka = Kafka { kafkaPtr :: RdKafkaTPtr, _kafkaConf :: KafkaConf} + +-- | Main pointer to Kafka topic, which is what we consume from or produce to +data KafkaTopic = KafkaTopic + RdKafkaTopicTPtr + Kafka -- Kept around to prevent garbage collection + KafkaTopicConf + +-- +-- Consumer +-- + +-- | Starting locations for a consumer +data KafkaOffset = + -- | Start reading from the beginning of the partition + KafkaOffsetBeginning + + -- | Start reading from the end + | KafkaOffsetEnd + + -- | Start reading from a specific location within the partition + | KafkaOffset Int64 + + -- | Start reading from the stored offset. See + -- + -- for offset store configuration. + | KafkaOffsetStored + +-- | Represents /received/ messages from a Kafka broker (i.e. used in a consumer) +data KafkaMessage = + KafkaMessage { + -- | Kafka partition this message was received from + messagePartition :: !Int + -- | Offset within the 'messagePartition' Kafka partition + , messageOffset :: !Int64 + -- | Contents of the message, as a 'ByteString' + , messagePayload :: !BS.ByteString + -- | Optional key of the message. 'Nothing' when the message + -- was enqueued without a key + , messageKey :: Maybe BS.ByteString + } + deriving (Eq, Show, Read, Typeable) + +-- +-- Producer +-- + +-- | Represents messages /to be enqueued/ onto a Kafka broker (i.e. used for a producer) +data KafkaProduceMessage = + -- | A message without a key, assigned to 'KafkaSpecifiedPartition' or 'KafkaUnassignedPartition' + KafkaProduceMessage + {-# UNPACK #-} !BS.ByteString -- message payload + + -- | A message with a key, assigned to a partition based on the key + | KafkaProduceKeyedMessage + {-# UNPACK #-} !BS.ByteString -- message key + {-# UNPACK #-} !BS.ByteString -- message payload + deriving (Eq, Show, Typeable) + +-- | Options for destination partition when enqueuing a message +data KafkaProducePartition = + -- | A specific partition in the topic + KafkaSpecifiedPartition {-# UNPACK #-} !Int -- the partition number of the topic + + -- | A random partition within the topic + | KafkaUnassignedPartition + +-- +-- Metadata +-- +-- | Metadata for all Kafka brokers +data KafkaMetadata = KafkaMetadata + { + -- | Broker metadata + brokers :: [KafkaBrokerMetadata] + -- | topic metadata + , topics :: [Either KafkaError KafkaTopicMetadata] + } + deriving (Eq, Show, Typeable) + +-- | Metadata for a specific Kafka broker +data KafkaBrokerMetadata = KafkaBrokerMetadata + { + -- | broker identifier + brokerId :: Int + -- | hostname for the broker + , brokerHost :: String + -- | port for the broker + , brokerPort :: Int + } + deriving (Eq, Show, Typeable) + +-- | Metadata for a specific topic +data KafkaTopicMetadata = KafkaTopicMetadata + { + -- | name of the topic + topicName :: String + -- | partition metadata + , topicPartitions :: [Either KafkaError KafkaPartitionMetadata] + } deriving (Eq, Show, Typeable) + +-- | Metadata for a specific partition +data KafkaPartitionMetadata = KafkaPartitionMetadata + { + -- | identifier for the partition + partitionId :: Int + + -- | broker leading this partition + , partitionLeader :: Int + + -- | replicas of the leader + , partitionReplicas :: [Int] + + -- | In-sync replica set, see + , partitionIsrs :: [Int] + } + deriving (Eq, Show, Typeable) + +-- +-- Helpers, exposed directly +-- + +-- | Log levels for the RdKafkaLibrary used in 'setKafkaLogLevel' +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 + +-- | Any Kafka errors +data KafkaError = + KafkaError String + | KafkaInvalidReturnValue + | KafkaBadSpecification String + | KafkaResponseError RdKafkaRespErrT + | KafkaInvalidConfigurationValue String + | KafkaUnknownConfigurationKey String + | KakfaBadConfiguration + deriving (Eq, Show, Typeable) + +instance Exception KafkaError diff --git a/tests/TestMain.hs b/tests/TestMain.hs --- a/tests/TestMain.hs +++ b/tests/TestMain.hs @@ -1,6 +1,7 @@ {-# LANGUAGE ScopedTypeVariables #-} module Main (main) where import Haskakafka +import Haskakafka.InternalSetup import Control.Exception import Control.Monad @@ -40,10 +41,8 @@ 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) +primeEOF :: KafkaTopic -> IO () +primeEOF kt = consumeMessage kt 0 1000 >> return () testmain :: IO () testmain = hspec $ do @@ -51,11 +50,6 @@ it "should be a valid version number" $ do rdKafkaVersionStr `shouldSatisfy` (=~"[0-9]+(.[0-9]+)+") - describe "Supported properties" $ do - it "should list supported properties" $ do - props <- supportedKafkaConfProperties - props `shouldSatisfy` (\x -> (length x) > 0) - describe "Kafka Configuration" $ do it "should allow dumping" $ do kConf <- newKafkaConf @@ -103,14 +97,13 @@ describe "Logging" $ do it "should allow setting of log level" $ getAddressTopic $ \a t -> do withKafkaConsumer [] [] a t 0 KafkaOffsetEnd $ \kafka _ -> do - setKafkaLogLevel kafka KafkaLogDebug + setLogLevel 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) + primeEOF topic perr <- withKafkaProducer [] [] a t $ \_ producerTopic -> do produceMessage producerTopic (KafkaSpecifiedPartition 0) message perr `shouldBe` Nothing @@ -124,7 +117,7 @@ let message = KafkaProduceKeyedMessage (C8.pack "key") (C8.pack "monkey around") withKafkaConsumer [] [] a t 0 KafkaOffsetEnd $ \_ topic -> do - shouldBeEmptyTopic topic + primeEOF topic perr <- withKafkaProducer [] [] a t $ \_ producerTopic -> do produceKeyedMessage producerTopic message perr `shouldBe` Nothing @@ -136,7 +129,7 @@ it "should be able to batch produce messages" $ getAddressTopic $ \a t -> do withKafkaConsumer [] [] a t 0 KafkaOffsetEnd $ \_ topic -> do - shouldBeEmptyTopic topic + primeEOF topic errs <- withKafkaProducer [] [] a t $ \_ producerTopic -> do produceMessageBatch producerTopic (KafkaSpecifiedPartition 0 ) sampleProduceMessages errs `shouldBe` [] @@ -150,12 +143,12 @@ it "should be able to batch consume messages" $ getAddressTopic $ \a t -> do withKafkaConsumer [] [] a t 0 KafkaOffsetEnd $ \_ topic -> do - shouldBeEmptyTopic topic + primeEOF topic errs <- withKafkaProducer [] [] a t $ \_ producerTopic -> do produceMessageBatch producerTopic (KafkaSpecifiedPartition 0 ) sampleProduceMessages errs `shouldBe` [] - et <- consumeMessageBatch topic 0 (1000) 3 + et <- consumeMessageBatch topic 0 (5000) 3 case et of (Left err) -> error $ show err (Right oms) -> do @@ -165,21 +158,19 @@ -- 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 + me <- fetchBrokerMetadata [] a 1000 return $ case me of (Left _) -> False (Right _) -> True main :: IO () main = do + a <- brokerAddress hasKafka <- checkForKafka if hasKafka then testmain - else error "\n\n\ + 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\ + \Haskakafka's tests require an operable Kafka broker running on " ++ a ++ "\n\ + \please follow the guide in Readme.md to set this up \n\ \*******************************************************************************\n"