diff --git a/example/Basic.hs b/example/Basic.hs new file mode 100644 --- /dev/null +++ b/example/Basic.hs @@ -0,0 +1,158 @@ +{-# LANGUAGE DeriveDataTypeable #-} +{-# LANGUAGE RankNTypes #-} +import Haskakafka + +import qualified Data.ByteString.Char8 as C8 +import Data.Maybe (fromMaybe) +import Control.Monad (forM_) + +import System.Console.CmdArgs +import Text.Show.Pretty (ppShow) + + +data BasicMode = Consumer | Producer | List | All + deriving (Data, Typeable, Show, Eq) + +data CArgs = CArgs + { brokers'CArgs :: Maybe String + , topic'CArgs :: Maybe String + , mode'CArgs :: BasicMode + , partition'CArgs :: Maybe Int + , pretty'CArgs :: Bool + } deriving (Data, Typeable, Show, Eq) + +producerExample :: + String -> String -> Int -> + (forall a. Show a => a -> String) -> IO () +producerExample brokerNames ourTopic partition showFn = 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 + brokerNames ourTopic + $ \kafka topic -> do + + -- Produce a single unkeyed message to partition + let message = KafkaProduceMessage samplePayload + _ <- produceMessage topic (KafkaSpecifiedPartition partition) message + + -- Produce a single keyed message + let keyMessage = KafkaProduceKeyedMessage (C8.pack "Key") samplePayload + _ <- produceKeyedMessage topic keyMessage + + -- We can also use the batch API for better performance + let numMessages = 9 + _ <- produceMessageBatch topic + KafkaUnassignedPartition + (replicate numMessages message) + + putStrLn "Done producing messages, here was our config: " + dumpConfFromKafka kafka >>= \d -> putStrLn $ "Kafka config: " ++ (showFn d) + dumpConfFromKafkaTopic topic >>= \d -> putStrLn $ "Topic config: " ++ (showFn d) + +consumerExample :: + String -> String -> Int -> + (forall a. Show a => a -> String) -> IO () +consumerExample brokerNames ourTopic partition showFn = do + let + -- Optionally, we can configure certain parameters for Kafka + kafkaConfig = [("socket.timeout.ms", "50000")] + topicConfig = [("request.timeout.ms", "50000")] + + -- withKafkaConsumer opens a consumer connection and starts consuming + withKafkaConsumer kafkaConfig topicConfig + brokerNames ourTopic + 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 = 3000 + me <- consumeMessage topic partition timeoutMs + case me of + (Left err) -> putStrLn $ "Uh oh, an error! " ++ (showFn 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! " ++ (showFn err) + (Right ms) -> forM_ ms $ \msg -> do + putStrLn $ "Woohoo, we got: " ++ (showFn msg) + + -- Be a little less noisy + setLogLevel kafka KafkaLogCrit + +metadataExample :: String -> (forall a. Show a => a -> String) -> IO () +metadataExample brokerNames showFn = do + -- we can also fetch metadata about our Kafka infrastructure + let timeoutMs = 1000 + emd <- fetchBrokerMetadata [] brokerNames timeoutMs + case emd of + (Left err) -> putStrLn $ "Uh oh, error time: " ++ (showFn err) + (Right md) -> putStrLn $ "Kafka metadata: " ++ (showFn md) + +runExample :: + BasicMode -> String -> String -> Int -> + (forall a. (Show a) => a -> String) -> IO () +runExample Consumer b t p pp = consumerExample b t p pp +runExample Producer b t p pp = producerExample b t p pp +runExample List b _ _ pp = metadataExample b pp +runExample All b t p pp = do + -- consumerExample b t p pp + producerExample b t p pp + metadataExample b pp + +parseExample :: CArgs -> IO () +parseExample (CArgs b t m p pp) = runExample + m + (fromMaybe "localhost:9092" b) + (fromMaybe "test_topic" t) + (fromMaybe 0 p) + (if pp then ppShow else show) + +cargs :: CArgs +cargs = CArgs + { brokers'CArgs = def + &= typ "" + &= help "Comma separated list in format :,:" + &= explicit + &= name "brokers" + &= name "b" + , topic'CArgs = def + &= typ "" + &= help "Topic to fetch / produce" + &= explicit + &= name "topic" + &= name "t" + , mode'CArgs = enum + [ Consumer &= help "Consumer mode" &= name "C" + , Producer &= help "Producer mode" &= name "P" + , List &= help "Metadata list mode" &= name "L" + , All &= help "Run producer, consumer, and metadata list" &= name "A" + ] + , partition'CArgs = def + &= typ "" + &= help "Partition (-1 for random partitioner when using producer)" + &= explicit + &= name "p" + , pretty'CArgs = def + &= help "Pretty print output" + &= explicit + &= name "pretty" + } &= help "Fetch metadata, produce, and consume a message" + &= program "basic example" + +main :: IO () +main = parseExample =<< cmdArgs cargs diff --git a/example/Simple.hs b/example/Simple.hs new file mode 100644 --- /dev/null +++ b/example/Simple.hs @@ -0,0 +1,75 @@ +import Haskakafka + +import qualified Data.ByteString.Char8 as C8 + +-- | Attempts to connect to kafka at localhost:9092. If you want to specify +main :: IO () +main = do + let + brokerNames = "localhost:9092" + ourTopic = "test_topic" + + -- 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 + brokerNames ourTopic + $ \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 keyMessage + + -- We can also use the batch API for better performance + _ <- produceMessageBatch topic KafkaUnassignedPartition + (replicate 9 message) + + 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 + brokerNames ourTopic + 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 + setLogLevel kafka KafkaLogCrit + + -- we can also fetch metadata about our Kafka infrastructure + let timeoutMs = 1000 + emd <- fetchBrokerMetadata [] brokerNames timeoutMs + case emd of + (Left err) -> putStrLn $ "Uh oh, error time: " ++ (show err) + (Right md) -> putStrLn $ "Kafka metadata: " ++ (show md) diff --git a/haskakafka.cabal b/haskakafka.cabal --- a/haskakafka.cabal +++ b/haskakafka.cabal @@ -1,9 +1,9 @@ name: haskakafka -version: 1.0.0 +version: 1.1.0 synopsis: Kafka bindings for Haskell description: Apache Kafka bindings backed by the librdkafka - C library. This implies full consumer and producer - support for Kafka 0.8.x. + C library. This implies full consumer and producer + support for Kafka 0.9.x. homepage: http://github.com/cosbynator/haskakafka license: MIT license-file: LICENSE @@ -21,7 +21,7 @@ Build-tools: c2hs build-depends: base >=4.6 && < 5 , bytestring - , containers + , containers , temporary , unix exposed-modules: @@ -30,14 +30,40 @@ Haskakafka.InternalRdKafkaEnum Haskakafka.InternalSetup Haskakafka.InternalTypes - Haskakafka.Example - other-modules: + Haskakafka.ConsumerExample + Haskakafka.Consumer + Haskakafka.Consumer.Internal.Convert + Haskakafka.Consumer.Internal.Types + Haskakafka.InternalShared + other-modules: hs-source-dirs: src default-language: Haskell2010 - ghc-options: -Wall + ghc-options: -Wall include-dirs: /usr/local/include/librdkafka - extra-lib-dirs: /usr/local/lib + extra-lib-dirs: /usr/local/lib extra-libraries: rdkafka + +executable simple + main-is: Simple.hs + hs-source-dirs: example + ghc-options: -Wall + default-language: Haskell2010 + build-depends: + base + , haskakafka + , bytestring + +executable basic + main-is: Basic.hs + hs-source-dirs: example + ghc-options: -Wall + default-language: Haskell2010 + build-depends: + base + , haskakafka + , bytestring + , cmdargs + , pretty-show test-suite tests type: exitcode-stdio-1.0 diff --git a/src/Haskakafka.hs b/src/Haskakafka.hs --- a/src/Haskakafka.hs +++ b/src/Haskakafka.hs @@ -1,4 +1,4 @@ -module Haskakafka +module Haskakafka ( fetchBrokerMetadata , withKafkaConsumer , consumeMessage @@ -48,112 +48,91 @@ ) where -import Haskakafka.InternalRdKafka -import Haskakafka.InternalRdKafkaEnum -import Haskakafka.InternalSetup -import Haskakafka.InternalTypes +import Haskakafka.InternalRdKafka +import Haskakafka.InternalRdKafkaEnum +import Haskakafka.InternalSetup +import Haskakafka.InternalShared +import Haskakafka.InternalTypes -import Control.Exception -import Control.Monad -import Foreign -import Foreign.C.Error -import Foreign.C.String -import Foreign.C.Types +import Control.Exception +import Control.Monad +import Foreign +import Foreign.C.Error +import Foreign.C.String +import Foreign.C.Types -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 Data.ByteString.Internal as BSI import qualified Haskakafka.InternalRdKafkaEnum as RDE +import qualified Haskakafka.InternalSetup as IS +import qualified Haskakafka.InternalTypes as IT +import Data.Either + -- | 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) + 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 = +startConsuming (KafkaTopic topicPtr _ _) partition offset = let trueOffset = case offset of KafkaOffsetBeginning -> (- 2) KafkaOffsetEnd -> (- 1) KafkaOffsetStored -> (- 1000) + KafkaOffsetInvalid -> (- 1001) KafkaOffset i -> i in throwOnError $ rdKafkaConsumeStart topicPtr partition trueOffset -- | 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 = +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) - - 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 -- | Consumes a single message from a Kafka topic, waiting up to a given timeout -consumeMessage :: KafkaTopic +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 -> - 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 - 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 $ Right $ KafkaMessage - (partition'RdKafkaMessageT s) - (offset'RdKafkaMessageT s) - payload - key + fromMessagePtr ptr -- | 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 +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 = +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 + ms <- if numMessages /= 0 then + 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 + withForeignPtr fptr $ \realPtr -> + rdKafkaMessageDestroy realPtr + if (err'RdKafkaMessageT storable) /= RdKafkaRespErrNoError then + return $ Left $ KafkaResponseError $ err'RdKafkaMessageT storable + else + return $ Right ret + else return [] + case lefts ms of + [] -> return $ Right $ rights ms + l -> return $ Left $ head l -- | Store a partition's offset in librdkafka's offset store. This function only needs to be called -- if auto.commit.enable is false. See @@ -161,12 +140,12 @@ storeOffset :: KafkaTopic -> Int -> Int -> IO (Maybe KafkaError) storeOffset (KafkaTopic topicPtr _ _) partition offset = do err <- rdKafkaOffsetStore topicPtr (fromIntegral partition) (fromIntegral offset) - case err of + case err of RdKafkaRespErrNoError -> return Nothing e -> return $ Just $ KafkaResponseError e -- | 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 +-- 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. @@ -177,8 +156,8 @@ withForeignPtr payloadFPtr $ \payloadPtr -> do let passedPayload = payloadPtr `plusPtr` payloadOffset - - handleProduceErr =<< + + handleProduceErr =<< rdKafkaProduce topicPtr (producePartitionInteger partition) copyMsgFlags passedPayload (fromIntegral payloadLength) nullPtr (CSize 0) nullPtr @@ -200,7 +179,7 @@ let passedPayload = payloadPtr `plusPtr` payloadOffset passedKey = keyPtr `plusPtr` keyOffset - handleProduceErr =<< + handleProduceErr =<< rdKafkaProduce topicPtr (producePartitionInteger KafkaUnassignedPartition) copyMsgFlags passedPayload (fromIntegral payloadLength) passedKey (fromIntegral keyLength) nullPtr @@ -217,44 +196,48 @@ batchPtrF <- newForeignPtr_ batchPtr numRet <- rdKafkaProduceBatch topicPtr partitionInt copyMsgFlags batchPtrF (length storables) if numRet == (length storables) then return [] - else do - errs <- mapM (\i -> return . err'RdKafkaMessageT =<< peekElemOff batchPtr i) + else do + errs <- mapM (\i -> return . err'RdKafkaMessageT =<< peekElemOff batchPtr i) [0..((fromIntegral $ length storables) - 1)] return [(m, KafkaResponseError e) | (m, e) <- (zip pms errs), e /= RdKafkaRespErrNoError] where partitionInt = (producePartitionInteger partition) produceMessageToMessage (KafkaProduceMessage bs) = do let (payloadFPtr, payloadOffset, payloadLength) = BSI.toForeignPtr bs - withForeignPtr payloadFPtr $ \payloadPtr -> do - let passedPayload = payloadPtr `plusPtr` payloadOffset - return $ RdKafkaMessageT - { err'RdKafkaMessageT = RdKafkaRespErrNoError - , partition'RdKafkaMessageT = fromIntegral partitionInt - , len'RdKafkaMessageT = payloadLength - , payload'RdKafkaMessageT = passedPayload - , offset'RdKafkaMessageT = 0 - , keyLen'RdKafkaMessageT = 0 - , key'RdKafkaMessageT = nullPtr - } + withForeignPtr topicPtr $ \ptrTopic -> do + withForeignPtr payloadFPtr $ \payloadPtr -> do + let passedPayload = payloadPtr `plusPtr` payloadOffset + return $ RdKafkaMessageT + { err'RdKafkaMessageT = RdKafkaRespErrNoError + , topic'RdKafkaMessageT = ptrTopic + , partition'RdKafkaMessageT = fromIntegral partitionInt + , len'RdKafkaMessageT = payloadLength + , payload'RdKafkaMessageT = passedPayload + , offset'RdKafkaMessageT = 0 + , keyLen'RdKafkaMessageT = 0 + , key'RdKafkaMessageT = nullPtr + } produceMessageToMessage (KafkaProduceKeyedMessage kbs bs) = do let (payloadFPtr, payloadOffset, payloadLength) = BSI.toForeignPtr bs (keyFPtr, keyOffset, keyLength) = BSI.toForeignPtr kbs - - withForeignPtr payloadFPtr $ \payloadPtr -> do - withForeignPtr keyFPtr $ \keyPtr -> do - let passedPayload = payloadPtr `plusPtr` payloadOffset - passedKey = keyPtr `plusPtr` keyOffset - return $ RdKafkaMessageT - { err'RdKafkaMessageT = RdKafkaRespErrNoError - , partition'RdKafkaMessageT = fromIntegral partitionInt - , len'RdKafkaMessageT = payloadLength - , payload'RdKafkaMessageT = passedPayload - , offset'RdKafkaMessageT = 0 - , keyLen'RdKafkaMessageT = keyLength - , key'RdKafkaMessageT = passedKey - } + withForeignPtr topicPtr $ \ptrTopic -> + withForeignPtr payloadFPtr $ \payloadPtr -> do + withForeignPtr keyFPtr $ \keyPtr -> do + let passedPayload = payloadPtr `plusPtr` payloadOffset + passedKey = keyPtr `plusPtr` keyOffset + return $ RdKafkaMessageT + { err'RdKafkaMessageT = RdKafkaRespErrNoError + , topic'RdKafkaMessageT = ptrTopic + , partition'RdKafkaMessageT = fromIntegral partitionInt + , len'RdKafkaMessageT = payloadLength + , payload'RdKafkaMessageT = passedPayload + , offset'RdKafkaMessageT = 0 + , keyLen'RdKafkaMessageT = keyLength + , key'RdKafkaMessageT = passedKey + } + -- | 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' @@ -266,7 +249,7 @@ -> (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 + bracket (do kafka <- newKafka RdKafkaProducer configOverrides addBrokers kafka brokerString @@ -277,7 +260,7 @@ (\(k, t) -> cb k t) -- | Connects to Kafka broker in consumer mode for a specific partition, --- taking a function that is fed with +-- 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. @@ -312,7 +295,7 @@ {-# INLINE handleProduceErr #-} handleProduceErr :: Int -> IO (Maybe KafkaError) -handleProduceErr (- 1) = getErrno >>= return . Just . kafkaRespErr +handleProduceErr (- 1) = getErrno >>= return . Just . kafkaRespErr handleProduceErr 0 = return $ Nothing handleProduceErr _ = return $ Just $ KafkaInvalidReturnValue @@ -327,35 +310,35 @@ getAllMetadata kafka timeout -- | Grabs all metadata from a given Kafka instance. -getAllMetadata :: Kafka +getAllMetadata :: Kafka -> Int -- ^ timeout in milliseconds (10^3 per second) -> IO (Either KafkaError KafkaMetadata) getAllMetadata k timeout = getMetadata k Nothing timeout -- | Grabs topic metadata from a given Kafka topic instance -getTopicMetadata :: Kafka - -> KafkaTopic +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 + case err of Left e -> return $ Left $ e - Right md -> case (topics md) of + Right md -> case (topics md) of [(Left e)] -> return $ Left e [(Right tmd)] -> return $ Right tmd _ -> 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 - err <- case mTopic of - Just (KafkaTopic kTopicPtr _ _) -> + err <- case mTopic of + Just (KafkaTopic kTopicPtr _ _) -> rdKafkaMetadata kPtr False kTopicPtr mdDblPtr timeout Nothing -> do nullTopic <- newForeignPtr_ nullPtr rdKafkaMetadata kPtr True nullTopic mdDblPtr timeout - case err of + case err of RdKafkaRespErrNoError -> do mdPtr <- peek mdDblPtr md <- peek mdPtr @@ -364,7 +347,7 @@ return $ Right $ retMd e -> return $ Left $ KafkaResponseError e - where + where constructMetadata md = do let nBrokers = (brokerCnt'RdKafkaMetadataT md) brokersPtr = (brokers'RdKafkaMetadataT md) @@ -377,7 +360,7 @@ constructBrokerMetadata bmd = do hostStr <- peekCString (host'RdKafkaMetadataBrokerT bmd) - return $ KafkaBrokerMetadata + return $ KafkaBrokerMetadata (id'RdKafkaMetadataBrokerT bmd) (hostStr) (port'RdKafkaMetadataBrokerT bmd) @@ -402,8 +385,8 @@ isrsPtr = (isrs'RdKafkaMetadataPartitionT pmd) replicas <- mapM (\i -> peekElemOff replicasPtr i) [0..((fromIntegral nReplicas) - 1)] isrs <- mapM (\i -> peekElemOff isrsPtr i) [0..((fromIntegral nIsrs) - 1)] - return $ Right $ KafkaPartitionMetadata - (id'RdKafkaMetadataPartitionT pmd) + return $ Right $ KafkaPartitionMetadata + (id'RdKafkaMetadataPartitionT pmd) (leader'RdKafkaMetadataPartitionT pmd) (map fromIntegral replicas) (map fromIntegral isrs) @@ -415,7 +398,7 @@ 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 +-- | 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 @@ -423,14 +406,3 @@ l <- outboundQueueLength k if l == 0 then return () else drainOutQueue k - -kafkaRespErr :: Errno -> KafkaError -kafkaRespErr (Errno num) = KafkaResponseError $ rdKafkaErrno2err (fromIntegral num) - -throwOnError :: IO (Maybe String) -> IO () -throwOnError action = do - m <- action - case m of - Just e -> throw $ KafkaError e - Nothing -> return () - diff --git a/src/Haskakafka/Consumer.hs b/src/Haskakafka/Consumer.hs new file mode 100644 --- /dev/null +++ b/src/Haskakafka/Consumer.hs @@ -0,0 +1,167 @@ +module Haskakafka.Consumer +( runConsumerConf +, runConsumer +, newKafkaConsumerConf +, newKafkaConsumer +, setRebalanceCallback +, assign +, subscribe +, pollMessage +, closeConsumer +, setOffsetStore + +-- Types +, CIT.ConsumerGroupId (..) +, CIT.TopicName (..) +, CIT.BrokersString (..) +, CIT.KafkaTopicPartition (..) +) +where + +import Control.Exception +import Foreign +import Haskakafka +import Haskakafka.Consumer.Internal.Convert +import Haskakafka.Consumer.Internal.Types +import Haskakafka.InternalRdKafka +import Haskakafka.InternalRdKafkaEnum +import Haskakafka.InternalSetup +import Haskakafka.InternalShared +import Haskakafka.InternalTypes + +import qualified Haskakafka.Consumer.Internal.Types as CIT + +-- | Runs high-level kafka consumer. +-- +-- A callback provided is expected to call 'pollMessage' when convenient. +runConsumerConf :: KafkaConf -- ^ Consumer config (see 'newKafkaConsumerConf') + -> BrokersString -- ^ Comma separated list of brokers with ports (e.g. @localhost:9092@) + -> [TopicName] -- ^ List of topics to be consumed + -> (Kafka -> IO (Either KafkaError ())) -- ^ A callback function to poll and handle messages + -> IO (Either KafkaError ()) +runConsumerConf c bs ts f = + bracket mkConsumer clConsumer runHandler + where + mkConsumer = do + kafka <- newKafkaConsumer bs c + _ <- setHlConsumer kafka + sErr <- subscribe kafka ts + return $ if hasError sErr + then Left (sErr, kafka) + else Right kafka + + clConsumer (Left (_, kafka)) = kafkaErrorToEither <$> closeConsumer kafka + clConsumer (Right kafka) = kafkaErrorToEither <$> closeConsumer kafka + + runHandler (Left (err, _)) = return $ Left err + runHandler (Right kafka) = f kafka + +-- | Runs high-level kafka consumer. +-- +-- A callback provided is expected to call 'pollMessage' when convenient. +runConsumer :: ConsumerGroupId -- ^ Consumer group id (a @group.id@ property of a kafka consumer) + -> ConfigOverrides -- ^ Extra kafka consumer parameters (see kafka documentation) + -> BrokersString -- ^ Comma separated list of brokers with ports (e.g. @localhost:9092@) + -> [TopicName] -- ^ List of topics to be consumed + -> (Kafka -> IO (Either KafkaError ())) -- ^ A callback function to poll and handle messages + -> IO (Either KafkaError ()) +runConsumer g c bs ts f = do + conf <- newKafkaConsumerConf g c + runConsumerConf conf bs ts f + +-- | Creates a new kafka configuration for a consumer with a specified 'ConsumerGroupId'. +newKafkaConsumerConf :: ConsumerGroupId -- ^ Consumer group id (a @group.id@ property of a kafka consumer) + -> ConfigOverrides -- ^ Extra kafka consumer parameters (see kafka documentation) + -> IO KafkaConf -- ^ Kafka configuration which can be altered before it is used in 'newKafkaConsumer' +newKafkaConsumerConf (ConsumerGroupId gid) conf = do + kc <- kafkaConf conf + setKafkaConfValue kc "group.id" gid + return kc + +-- | Creates a new kafka consumer +newKafkaConsumer :: BrokersString -- ^ Comma separated list of brokers with ports (e.g. @localhost:9092@) + -> KafkaConf -- ^ Kafka configuration for a consumer (see 'newKafkaConsumerConf') + -> IO Kafka -- ^ Kafka instance +newKafkaConsumer (BrokersString bs) conf = do + kafka <- newKafkaPtr RdKafkaConsumer conf + addBrokers kafka bs + return kafka + +-- | Sets a callback that is called when rebalance is needed. +-- +-- Callback implementations suppose to watch for 'KafkaResponseError' 'RdKafkaRespErrAssignPartitions' and +-- for 'KafkaResponseError' 'RdKafkaRespErrRevokePartitions'. Other error codes are not expected and would indicate +-- something really bad happening in a system, or bugs in @librdkafka@ itself. +-- +-- A callback is expected to call 'assign' according to the error code it receives. +-- +-- * When 'RdKafkaRespErrAssignPartitions' happens 'assign' should be called with all the partitions it was called with. +-- It is OK to alter partitions offsets before calling 'assign'. +-- +-- * When 'RdKafkaRespErrRevokePartitions' happens 'assign' should be called with an empty list of partitions. +setRebalanceCallback :: KafkaConf + -> (Kafka -> KafkaError -> [KafkaTopicPartition] -> IO ()) + -> IO () +setRebalanceCallback (KafkaConf conf) callback = rdKafkaConfSetRebalanceCb conf realCb + where + realCb :: Ptr RdKafkaT -> RdKafkaRespErrT -> Ptr RdKafkaTopicPartitionListT -> Ptr Word8 -> IO () + realCb rk err pl _ = do + rk' <- newForeignPtr_ rk + pl' <- peek pl + ps <- fromNativeTopicPartitionList pl' + callback (Kafka rk' (KafkaConf conf)) (KafkaResponseError err) ps + +-- | Assigns specified partitions to a current consumer. +-- Assigning an empty list means unassigning from all partitions that are currently assigned. +-- See 'setRebalanceCallback' for more details. +assign :: Kafka -> [KafkaTopicPartition] -> IO KafkaError +assign (Kafka k _) ps = + let pl = if null ps + then newForeignPtr_ nullPtr + else toNativeTopicPartitionList ps + in KafkaResponseError <$> (pl >>= rdKafkaAssign k) + +-- | Subscribes to a given list of topics. +-- +-- Wildcard (regex) topics are supported by the librdkafka assignor: +-- any topic name in the topics list that is prefixed with @^@ will +-- be regex-matched to the full list of topics in the cluster and matching +-- topics will be added to the subscription list. +subscribe :: Kafka -> [TopicName] -> IO KafkaError +subscribe (Kafka k _) ts = do + pl <- newRdKafkaTopicPartitionListT (length ts) + mapM_ (\(TopicName t) -> rdKafkaTopicPartitionListAdd pl t (-1)) ts + KafkaResponseError <$> rdKafkaSubscribe k pl + +-- | Closes the consumer and destroys it. +closeConsumer :: Kafka -> IO KafkaError +closeConsumer (Kafka k _) = KafkaResponseError <$> rdKafkaConsumerClose k + +----------------------------------------------------------------------------- +setTopicValue :: KafkaTopic -> String -> String -> IO () +setTopicValue (KafkaTopic _ _ conf) = setKafkaTopicConfValue conf + +pollMessage :: Kafka + -> Int -- ^ the timeout, in milliseconds (@10^3@ per second) + -> IO (Either KafkaError KafkaMessage) -- ^ Left on error or timeout, right for success +pollMessage (Kafka k _) timeout = + rdKafkaConsumerPoll k (fromIntegral timeout) >>= fromMessagePtr + +-- | Redirects 'consumeMessage' to poll. Implementation details. +setHlConsumer :: Kafka -> IO KafkaError +setHlConsumer (Kafka k _) = KafkaResponseError <$> rdKafkaPollSetConsumer k + +-- | Sets the offset store for a specified topic. +-- @librdkafka@ supports both @broker@ and @file@ but it seems that consumers with groups +-- can only support @broker@. Which is good and enough. +setOffsetStore :: KafkaTopic -> OffsetStoreMethod -> IO () +setOffsetStore t o = + let setValue = setTopicValue t + in case o of + OffsetStoreBroker -> + setValue "offset.store.method" "broker" + + OffsetStoreFile path sync -> do + setValue "offset.store.method" "file" + setValue "offset.store.file" path + setValue "offset.store.sync.interval.ms" (show $ offsetSyncToInt sync) diff --git a/src/Haskakafka/Consumer/Internal/Convert.hs b/src/Haskakafka/Consumer/Internal/Convert.hs new file mode 100644 --- /dev/null +++ b/src/Haskakafka/Consumer/Internal/Convert.hs @@ -0,0 +1,71 @@ +module Haskakafka.Consumer.Internal.Convert + +where + +import Control.Monad +import Data.Int +import Foreign +import Foreign.C.String +import Haskakafka.Consumer.Internal.Types +import Haskakafka.InternalRdKafka +import Haskakafka.InternalTypes + +-- | Converts offsets sync policy to integer (the way Kafka understands it): +-- +-- * @OffsetSyncDisable == -1@ +-- +-- * @OffsetSyncImmediate == 0@ +-- +-- * @OffsetSyncInterval ms == ms@ +offsetSyncToInt :: OffsetStoreSync -> Int +offsetSyncToInt sync = + case sync of + OffsetSyncDisable -> -1 + OffsetSyncImmediate -> 0 + OffsetSyncInterval ms -> ms +{-# INLINE offsetSyncToInt #-} + +offsetToInt64 :: KafkaOffset -> Int64 +offsetToInt64 o = case o of + KafkaOffsetBeginning -> -2 + KafkaOffsetEnd -> -1 + KafkaOffset off -> off + KafkaOffsetStored -> -1000 + KafkaOffsetInvalid -> -1001 +{-# INLINE offsetToInt64 #-} + +int64ToOffset :: Int64 -> KafkaOffset +int64ToOffset o + | o == -2 = KafkaOffsetBeginning + | o == -1 = KafkaOffsetEnd + | o == -1000 = KafkaOffsetStored + | o >= 0 = KafkaOffset o + | otherwise = KafkaOffsetInvalid +{-# INLINE int64ToOffset #-} + +fromNativeTopicPartitionList :: RdKafkaTopicPartitionListT -> IO [KafkaTopicPartition] +fromNativeTopicPartitionList pl = + let count = cnt'RdKafkaTopicPartitionListT pl + elems = elems'RdKafkaTopicPartitionListT pl + in mapM (peekElemOff elems >=> toPart) [0..(fromIntegral count - 1)] + where + toPart :: RdKafkaTopicPartitionT -> IO KafkaTopicPartition + toPart p = do + topic <- peekCString $ topic'RdKafkaTopicPartitionT p + return KafkaTopicPartition { + ktpTopicName = TopicName topic, + ktpPartition = partition'RdKafkaTopicPartitionT p, + ktpOffset = int64ToOffset $ offset'RdKafkaTopicPartitionT p + } + +toNativeTopicPartitionList :: [KafkaTopicPartition] -> IO RdKafkaTopicPartitionListTPtr +toNativeTopicPartitionList ps = do + pl <- newRdKafkaTopicPartitionListT (length ps) + mapM_ (\p -> do + let TopicName tn = ktpTopicName p + tp = ktpPartition p + to = offsetToInt64 $ ktpOffset p + _ <- rdKafkaTopicPartitionListAdd pl tn tp + rdKafkaTopicPartitionListSetOffset pl tn tp to) ps + return pl + diff --git a/src/Haskakafka/Consumer/Internal/Types.hs b/src/Haskakafka/Consumer/Internal/Types.hs new file mode 100644 --- /dev/null +++ b/src/Haskakafka/Consumer/Internal/Types.hs @@ -0,0 +1,38 @@ +module Haskakafka.Consumer.Internal.Types + +where + +import Haskakafka.InternalTypes + +newtype ConsumerGroupId = ConsumerGroupId String deriving (Show, Eq) + +-- | Topic name to be consumed +-- +-- Wildcard (regex) topics are supported by the librdkafka assignor: +-- any topic name in the topics list that is prefixed with @^@ will +-- be regex-matched to the full list of topics in the cluster and matching +-- topics will be added to the subscription list. +newtype TopicName = + TopicName String -- ^ a simple topic name or a regex if started with @^@ + deriving (Show, Eq) + +-- | Comma separated broker:port string (e.g. @broker1:9092,broker2:9092@) +newtype BrokersString = BrokersString String deriving (Show, Eq) + +-- | Indicates how offsets are to be synced to disk +data OffsetStoreSync = + OffsetSyncDisable -- ^ Do not sync offsets (in Kafka: -1) + | OffsetSyncImmediate -- ^ Sync immediately after each offset commit (in Kafka: 0) + | OffsetSyncInterval Int -- ^ Sync after specified interval in millis + +-- | Indicates the method of storing the offsets +data OffsetStoreMethod = + OffsetStoreBroker -- ^ Offsets are stored in Kafka broker (preferred) + | OffsetStoreFile FilePath OffsetStoreSync -- ^ Offsets are stored in a file (and synced to disk according to the sync policy) + +-- | Kafka topic partition structure +data KafkaTopicPartition = KafkaTopicPartition + { ktpTopicName :: TopicName + , ktpPartition :: Int + , ktpOffset :: KafkaOffset } deriving (Show, Eq) + diff --git a/src/Haskakafka/ConsumerExample.hs b/src/Haskakafka/ConsumerExample.hs new file mode 100644 --- /dev/null +++ b/src/Haskakafka/ConsumerExample.hs @@ -0,0 +1,62 @@ +module Haskakafka.ConsumerExample + +where + +import Control.Arrow ((&&&)) +import Haskakafka +import Haskakafka.Consumer +import Haskakafka.InternalRdKafkaEnum + +iterator :: [Integer] +iterator = [0 .. 20] + +runConsumerExample :: IO () +runConsumerExample = do + res <- runConsumer + (ConsumerGroupId "test_group") + [] + (BrokersString "localhost:9092") + [TopicName "^hl-test*"] + processMessages + print $ show res + +consumerExample :: IO () +consumerExample = do + print "creating kafka conf" + conf <- newKafkaConsumerConf (ConsumerGroupId "test_group") [] + + -- unnecessary, demo only + setRebalanceCallback conf printingRebalanceCallback + + res <- runConsumerConf + conf + (BrokersString "localhost:9092") + [TopicName "^hl-test*"] + processMessages + + print $ show res + +------------------------------------------------------------------- +processMessages :: Kafka -> IO (Either KafkaError ()) +processMessages kafka = do + mapM_ (\_ -> do + msg1 <- pollMessage kafka 1000 + print $ show msg1) iterator + return $ Right () + +printingRebalanceCallback :: Kafka -> KafkaError -> [KafkaTopicPartition] -> IO () +printingRebalanceCallback k e ps = do + print $ show e + print "partitions: " + mapM_ (print . show . (ktpTopicName &&& ktpPartition &&& ktpOffset)) ps + case e of + KafkaResponseError RdKafkaRespErrAssignPartitions -> do + err <- assign k ps + print $ "Assign result: " ++ show err + KafkaResponseError RdKafkaRespErrRevokePartitions -> do + err <- assign k [] + print $ "Revoke result: " ++ show err + x -> + print "UNKNOWN (and unlikely!)" >> print (show x) + + diff --git a/src/Haskakafka/Example.hs b/src/Haskakafka/Example.hs deleted file mode 100644 --- a/src/Haskakafka/Example.hs +++ /dev/null @@ -1,71 +0,0 @@ -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 - 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/InternalRdKafka.chs b/src/Haskakafka/InternalRdKafka.chs --- a/src/Haskakafka/InternalRdKafka.chs +++ b/src/Haskakafka/InternalRdKafka.chs @@ -3,16 +3,15 @@ module Haskakafka.InternalRdKafka where -import Control.Applicative +--import Control.Applicative import Control.Monad import Data.Word -import Foreign hiding (unsafePerformIO) +import Foreign 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 @@ -21,7 +20,7 @@ type CInt64T = {#type int64_t #} type CInt32T = {#type int32_t #} -{#pointer *FILE as CFilePtr -> CFile #} +{#pointer *FILE as CFilePtr -> CFile #} {#pointer *size_t as CSizePtr -> CSize #} type Word8Ptr = Ptr Word8 @@ -53,7 +52,7 @@ kafkaErrnoString :: IO (String) kafkaErrnoString = do - (Errno num) <- getErrno + (Errno num) <- getErrno return $ rdKafkaErr2str $ rdKafkaErrno2err (fromIntegral num) -- Kafka Pointer Types @@ -61,16 +60,69 @@ {#pointer *rd_kafka_conf_t as RdKafkaConfTPtr foreign -> RdKafkaConfT #} data RdKafkaTopicConfT -{#pointer *rd_kafka_topic_conf_t as RdKafkaTopicConfTPtr foreign -> RdKafkaTopicConfT #} +{#pointer *rd_kafka_topic_conf_t as RdKafkaTopicConfTPtr foreign -> RdKafkaTopicConfT #} data RdKafkaT {#pointer *rd_kafka_t as RdKafkaTPtr foreign -> RdKafkaT #} +data RdKafkaTopicPartitionT = RdKafkaTopicPartitionT + { topic'RdKafkaTopicPartitionT :: CString + , partition'RdKafkaTopicPartitionT :: Int + , offset'RdKafkaTopicPartitionT :: Int64 + , metadata'RdKafkaTopicPartitionT :: Word8Ptr + , metadataSize'RdKafkaTopicPartitionT :: Int + , opaque'RdKafkaTopicPartitionT :: Word8Ptr + , err'RdKafkaTopicPartitionT :: RdKafkaRespErrT + } deriving (Show, Eq) + +instance Storable RdKafkaTopicPartitionT where + alignment _ = {#alignof rd_kafka_topic_partition_t#} + sizeOf _ = {#sizeof rd_kafka_topic_partition_t#} + peek p = RdKafkaTopicPartitionT + <$> liftM id ({#get rd_kafka_topic_partition_t->topic #} p) + <*> liftM fromIntegral ({#get rd_kafka_topic_partition_t->partition #} p) + <*> liftM fromIntegral ({#get rd_kafka_topic_partition_t->offset #} p) + <*> liftM castPtr ({#get rd_kafka_topic_partition_t->metadata #} p) + <*> liftM fromIntegral ({#get rd_kafka_topic_partition_t->metadata_size #} p) + <*> liftM castPtr ({#get rd_kafka_topic_partition_t->opaque #} p) + <*> liftM cIntToEnum ({#get rd_kafka_topic_partition_t->err #} p) + poke p x = do + {#set rd_kafka_topic_partition_t.topic#} p (id $ topic'RdKafkaTopicPartitionT x) + {#set rd_kafka_topic_partition_t.partition#} p (fromIntegral $ partition'RdKafkaTopicPartitionT x) + {#set rd_kafka_topic_partition_t.offset#} p (fromIntegral $ offset'RdKafkaTopicPartitionT x) + {#set rd_kafka_topic_partition_t.metadata#} p (castPtr $ metadata'RdKafkaTopicPartitionT x) + {#set rd_kafka_topic_partition_t.metadata_size#} p (fromIntegral $ metadataSize'RdKafkaTopicPartitionT x) + {#set rd_kafka_topic_partition_t.opaque#} p (castPtr $ opaque'RdKafkaTopicPartitionT x) + {#set rd_kafka_topic_partition_t.err#} p (enumToCInt $ err'RdKafkaTopicPartitionT x) + +{#pointer *rd_kafka_topic_partition_t as RdKafkaTopicPartitionTPtr foreign -> RdKafkaTopicPartitionT #} + +data RdKafkaTopicPartitionListT = RdKafkaTopicPartitionListT + { cnt'RdKafkaTopicPartitionListT :: Int + , size'RdKafkaTopicPartitionListT :: Int + , elems'RdKafkaTopicPartitionListT :: Ptr RdKafkaTopicPartitionT + } deriving (Show, Eq) + +{#pointer *rd_kafka_topic_partition_list_t as RdKafkaTopicPartitionListTPtr foreign -> RdKafkaTopicPartitionListT #} + +instance Storable RdKafkaTopicPartitionListT where + alignment _ = {#alignof rd_kafka_topic_partition_list_t#} + sizeOf _ = {#sizeof rd_kafka_topic_partition_list_t #} + peek p = RdKafkaTopicPartitionListT + <$> liftM fromIntegral ({#get rd_kafka_topic_partition_list_t->cnt #} p) + <*> liftM fromIntegral ({#get rd_kafka_topic_partition_list_t->size #} p) + <*> liftM castPtr ({#get rd_kafka_topic_partition_list_t->elems #} p) + poke p x = do + {#set rd_kafka_topic_partition_list_t.cnt#} p (fromIntegral $ cnt'RdKafkaTopicPartitionListT x) + {#set rd_kafka_topic_partition_list_t.size#} p (fromIntegral $ size'RdKafkaTopicPartitionListT x) + {#set rd_kafka_topic_partition_list_t.elems#} p (castPtr $ elems'RdKafkaTopicPartitionListT x) + data RdKafkaTopicT {#pointer *rd_kafka_topic_t as RdKafkaTopicTPtr foreign -> RdKafkaTopicT #} -data RdKafkaMessageT = RdKafkaMessageT +data RdKafkaMessageT = RdKafkaMessageT { err'RdKafkaMessageT :: RdKafkaRespErrT + , topic'RdKafkaMessageT :: Ptr RdKafkaTopicT , partition'RdKafkaMessageT :: Int , len'RdKafkaMessageT :: Int , keyLen'RdKafkaMessageT :: Int @@ -79,26 +131,28 @@ , 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) + <$> liftM cIntToEnum ({#get rd_kafka_message_t->err #} p) + <*> liftM castPtr ({#get rd_kafka_message_t->rkt #} 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) + {#set rd_kafka_message_t.err#} p (enumToCInt $ err'RdKafkaMessageT x) + {#set rd_kafka_message_t.rkt#} p (castPtr $ topic'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 #} @@ -113,7 +167,7 @@ instance Storable RdKafkaMetadataBrokerT where alignment _ = {#alignof rd_kafka_metadata_broker_t#} - sizeOf _ = {#sizeof rd_kafka_metadata_broker_t#} + sizeOf _ = 24 peek p = RdKafkaMetadataBrokerT <$> liftM fromIntegral ({#get rd_kafka_metadata_broker_t->id #} p) <*> liftM id ({#get rd_kafka_metadata_broker_t->host #} p) @@ -155,13 +209,13 @@ instance Storable RdKafkaMetadataTopicT where alignment _ = {#alignof rd_kafka_metadata_topic_t#} - sizeOf _ = {#sizeof rd_kafka_metadata_topic_t #} + sizeOf _ = 32 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 + poke _ _ = undefined {#pointer *rd_kafka_metadata_topic_t as RdKafkaMetadataTopicTPtr -> RdKafkaMetadataTopicT #} @@ -182,13 +236,392 @@ <*> 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 + poke _ _ = undefined {#pointer *rd_kafka_metadata_t as RdKafkaMetadataTPtr foreign -> RdKafkaMetadataT #} +------------------------------------------------------------------------------------------------- +---- Partitions +{#fun unsafe rd_kafka_topic_partition_list_new as ^ + {`Int'} -> `RdKafkaTopicPartitionListTPtr' #} + +foreign import ccall unsafe "rdkafka.h &rd_kafka_topic_partition_list_destroy" + rdKafkaTopicPartitionListDestroy :: FunPtr (Ptr RdKafkaTopicPartitionListT -> IO ()) + +newRdKafkaTopicPartitionListT :: Int -> IO RdKafkaTopicPartitionListTPtr +newRdKafkaTopicPartitionListT size = do + ret <- rdKafkaTopicPartitionListNew size + addForeignPtrFinalizer rdKafkaTopicPartitionListDestroy ret + return ret + +{# fun unsafe rd_kafka_topic_partition_list_add as ^ + {`RdKafkaTopicPartitionListTPtr', `String', `Int'} -> `RdKafkaTopicPartitionTPtr' #} + +{# fun unsafe rd_kafka_topic_partition_list_add_range as ^ + {`RdKafkaTopicPartitionListTPtr', `String', `Int', `Int'} -> `()' #} + +{# fun unsafe rd_kafka_topic_partition_list_copy as ^ + {`RdKafkaTopicPartitionListTPtr'} -> `RdKafkaTopicPartitionListTPtr' #} + +copyRdKafkaTopicPartitionList :: RdKafkaTopicPartitionListTPtr -> IO RdKafkaTopicPartitionListTPtr +copyRdKafkaTopicPartitionList pl = do + cp <- rdKafkaTopicPartitionListCopy pl + addForeignPtrFinalizer rdKafkaTopicPartitionListDestroy cp + return cp + +{# fun unsafe rd_kafka_topic_partition_list_set_offset as ^ + {`RdKafkaTopicPartitionListTPtr', `String', `Int', `Int64'} + -> `RdKafkaRespErrT' cIntToEnum #} + +---- Rebalance Callback +type RdRebalanceCallback' = Ptr RdKafkaT -> CInt -> Ptr RdKafkaTopicPartitionListT -> Ptr Word8 -> IO () +type RdRebalanceCallback = Ptr RdKafkaT -> RdKafkaRespErrT -> Ptr RdKafkaTopicPartitionListT -> Ptr Word8 -> IO () + +foreign import ccall safe "wrapper" + mkRebalanceCallback :: RdRebalanceCallback' -> IO (FunPtr RdRebalanceCallback') + +foreign import ccall safe "rd_kafka.h rd_kafka_conf_set_rebalance_cb" + rdKafkaConfSetRebalanceCb' :: + Ptr RdKafkaConfT + -> FunPtr RdRebalanceCallback' + -> IO () + +rdKafkaConfSetRebalanceCb :: RdKafkaConfTPtr -> RdRebalanceCallback -> IO () +rdKafkaConfSetRebalanceCb conf cb = do + cb' <- mkRebalanceCallback (\k e p o -> cb k (cIntToEnum e) p o) + withForeignPtr conf $ \c -> rdKafkaConfSetRebalanceCb' c cb' + return () + +---- Delivery Callback +type DeliveryCallback = Ptr RdKafkaT -> Ptr RdKafkaMessageT -> Word8Ptr -> IO () + +foreign import ccall safe "wrapper" + mkDeliveryCallback :: DeliveryCallback -> IO (FunPtr DeliveryCallback) + +foreign import ccall unsafe "rd_kafka.h rd_kafka_conf_set_dr_msg_cb" + rdKafkaConfSetDrMsgCb' :: Ptr RdKafkaConfT -> FunPtr DeliveryCallback -> IO () + +rdKafkaConfSetDrMsgCb :: RdKafkaConfTPtr -> DeliveryCallback -> IO () +rdKafkaConfSetDrMsgCb conf cb = do + cb' <- mkDeliveryCallback cb + withForeignPtr conf $ \c -> rdKafkaConfSetDrMsgCb' c cb' + return () + +---- Consume Callback +type ConsumeCallback = Ptr RdKafkaMessageT -> Word8Ptr -> IO () + +foreign import ccall safe "wrapper" + mkConsumeCallback :: ConsumeCallback -> IO (FunPtr ConsumeCallback) + +foreign import ccall unsafe "rd_kafka.h rd_kafka_conf_set_consume_cb" + rdKafkaConfSetConsumeCb' :: Ptr RdKafkaConfT -> FunPtr ConsumeCallback -> IO () + +rdKafkaConfSetConsumeCb :: RdKafkaConfTPtr -> ConsumeCallback -> IO () +rdKafkaConfSetConsumeCb conf cb = do + cb' <- mkConsumeCallback cb + withForeignPtr conf $ \c -> rdKafkaConfSetConsumeCb' c cb' + return () + +---- Offset Commit Callback +type OffsetCommitCallback' = Ptr RdKafkaT -> CInt -> Ptr RdKafkaTopicPartitionListT -> Word8Ptr -> IO () +type OffsetCommitCallback = Ptr RdKafkaT -> RdKafkaRespErrT -> Ptr RdKafkaTopicPartitionListT -> Word8Ptr -> IO () + +foreign import ccall safe "wrapper" + mkOffsetCommitCallback :: OffsetCommitCallback' -> IO (FunPtr OffsetCommitCallback') + +foreign import ccall unsafe "rd_kafka.h rd_kafka_conf_set_offset_commit_cb" + rdKafkaConfSetOffsetCommitCb' :: Ptr RdKafkaConfT -> FunPtr OffsetCommitCallback' -> IO () + +rdKafkaConfSetOffsetCommitCb :: RdKafkaConfTPtr -> OffsetCommitCallback -> IO () +rdKafkaConfSetOffsetCommitCb conf cb = do + cb' <- mkOffsetCommitCallback (\k e p o -> cb k (cIntToEnum e) p o) + withForeignPtr conf $ \c -> rdKafkaConfSetOffsetCommitCb' c cb' + return () + +---- Throttle Callback +type ThrottleCallback = Ptr RdKafkaT -> CString -> Int -> Int -> Word8Ptr -> IO () + +foreign import ccall safe "wrapper" + mkThrottleCallback :: ThrottleCallback -> IO (FunPtr ThrottleCallback) + +foreign import ccall unsafe "rd_kafka.h rd_kafka_conf_set_throttle_cb" + rdKafkaConfSetThrottleCb' :: Ptr RdKafkaConfT -> FunPtr ThrottleCallback -> IO () + +rdKafkaConfSetThrottleCb :: RdKafkaConfTPtr -> ThrottleCallback -> IO () +rdKafkaConfSetThrottleCb conf cb = do + cb' <- mkThrottleCallback cb + withForeignPtr conf $ \c -> rdKafkaConfSetThrottleCb' c cb' + return () + +---- Stats Callback +type StatsCallback = Ptr RdKafkaT -> CString -> CSize -> Word8Ptr -> IO () + +foreign import ccall safe "wrapper" + mkStatsCallback :: StatsCallback -> IO (FunPtr StatsCallback) + +foreign import ccall unsafe "rd_kafka.h rd_kafka_conf_set_stats_cb" + rdKafkaConfSetStatsCb' :: Ptr RdKafkaConfT -> FunPtr StatsCallback -> IO () + +rdKafkaConfSetStatsCb :: RdKafkaConfTPtr -> StatsCallback -> IO () +rdKafkaConfSetStatsCb conf cb = do + cb' <- mkStatsCallback cb + withForeignPtr conf $ \c -> rdKafkaConfSetStatsCb' c cb' + return () + +---- Socket Callback +type SocketCallback = Int -> Int -> Int -> Word8Ptr -> IO () + +foreign import ccall safe "wrapper" + mkSocketCallback :: SocketCallback -> IO (FunPtr SocketCallback) + +foreign import ccall unsafe "rd_kafka.h rd_kafka_conf_set_socket_cb" + rdKafkaConfSetSocketCb' :: Ptr RdKafkaConfT -> FunPtr SocketCallback -> IO () + +rdKafkaConfSetSocketCb :: RdKafkaConfTPtr -> SocketCallback -> IO () +rdKafkaConfSetSocketCb conf cb = do + cb' <- mkSocketCallback cb + withForeignPtr conf $ \c -> rdKafkaConfSetSocketCb' c cb' + return () + +{#fun unsafe rd_kafka_conf_set_opaque as ^ + {`RdKafkaConfTPtr', castPtr `Word8Ptr'} -> `()' #} + +{#fun unsafe rd_kafka_opaque as ^ + {`RdKafkaTPtr'} -> `Word8Ptr' castPtr #} + +{#fun unsafe rd_kafka_conf_set_default_topic_conf as ^ + {`RdKafkaConfTPtr', `RdKafkaTopicConfTPtr'} -> `()' #} + +---- Partitioner Callback +type PartitionerCallback = + Ptr RdKafkaTopicTPtr + -> Word8Ptr -- keydata + -> Int -- keylen + -> Int -- partition_cnt + -> Word8Ptr -- topic_opaque + -> Word8Ptr -- msg_opaque + -> IO Int + +foreign import ccall safe "wrapper" + mkPartitionerCallback :: PartitionerCallback -> IO (FunPtr PartitionerCallback) + +foreign import ccall unsafe "rd_kafka.h rd_kafka_topic_conf_set_partitioner_cb" + rdKafkaTopicConfSetPartitionerCb' :: Ptr RdKafkaTopicConfT -> FunPtr PartitionerCallback -> IO () + +rdKafkaTopicConfSetPartitionerCb :: RdKafkaTopicConfTPtr -> PartitionerCallback -> IO () +rdKafkaTopicConfSetPartitionerCb conf cb = do + cb' <- mkPartitionerCallback cb + withForeignPtr conf $ \c -> rdKafkaTopicConfSetPartitionerCb' c cb' + return () + +---- Partition + +{#fun unsafe rd_kafka_topic_partition_available as ^ + {`RdKafkaTopicTPtr', cIntConv `CInt32T'} -> `Int' #} + +{#fun unsafe rd_kafka_msg_partitioner_random as ^ + { `RdKafkaTopicTPtr' + , castPtr `Word8Ptr' + , cIntConv `CSize' + , cIntConv `CInt32T' + , castPtr `Word8Ptr' + , castPtr `Word8Ptr'} + -> `CInt32T' cIntConv #} + +{#fun unsafe rd_kafka_msg_partitioner_consistent as ^ + { `RdKafkaTopicTPtr' + , castPtr `Word8Ptr' + , cIntConv `CSize' + , cIntConv `CInt32T' + , castPtr `Word8Ptr' + , castPtr `Word8Ptr'} + -> `CInt32T' cIntConv #} + +{#fun unsafe rd_kafka_msg_partitioner_consistent_random as ^ + { `RdKafkaTopicTPtr' + , castPtr `Word8Ptr' + , cIntConv `CSize' + , cIntConv `CInt32T' + , castPtr `Word8Ptr' + , castPtr `Word8Ptr'} + -> `CInt32T' cIntConv #} + +---- Poll / Yield + +{#fun unsafe rd_kafka_yield as ^ + {`RdKafkaTPtr'} -> `()' #} + +---- Pause / Resume +{#fun unsafe rd_kafka_pause_partitions as ^ + {`RdKafkaTPtr', `RdKafkaTopicPartitionListTPtr'} -> `RdKafkaRespErrT' cIntToEnum #} + +{#fun unsafe rd_kafka_resume_partitions as ^ + {`RdKafkaTPtr', `RdKafkaTopicPartitionListTPtr'} -> `RdKafkaRespErrT' cIntToEnum #} + +---- QUEUE +data RdKafkaQueueT +{#pointer *rd_kafka_queue_t as RdKafkaQueueTPtr foreign -> RdKafkaQueueT #} + +{#fun unsafe rd_kafka_queue_new as ^ + {`RdKafkaTPtr'} -> `RdKafkaQueueTPtr' #} + +foreign import ccall unsafe "rdkafka.h &rd_kafka_queue_destroy" + rdKafkaQueueDestroy :: FunPtr (Ptr RdKafkaQueueT -> IO ()) + +newRdKafkaQueue :: RdKafkaTPtr -> IO RdKafkaQueueTPtr +newRdKafkaQueue k = do + q <- rdKafkaQueueNew k + addForeignPtrFinalizer rdKafkaQueueDestroy q + return q +------------------------------------------------------------------------------------------------- +---- High-level KafkaConsumer + +{#fun unsafe rd_kafka_subscribe as ^ + {`RdKafkaTPtr', `RdKafkaTopicPartitionListTPtr'} + -> `RdKafkaRespErrT' cIntToEnum #} + +{#fun unsafe rd_kafka_unsubscribe as ^ + {`RdKafkaTPtr'} + -> `RdKafkaRespErrT' cIntToEnum #} + +{#fun unsafe rd_kafka_subscription as ^ + {`RdKafkaTPtr', castPtr `Ptr (Ptr RdKafkaTopicPartitionListT)'} + -> `RdKafkaRespErrT' cIntToEnum #} + +{#fun unsafe rd_kafka_consumer_poll as ^ + {`RdKafkaTPtr', `Int'} -> `RdKafkaMessageTPtr' #} + +{#fun unsafe rd_kafka_consumer_close as ^ + {`RdKafkaTPtr'} -> `RdKafkaRespErrT' cIntToEnum #} + +{#fun unsafe rd_kafka_poll_set_consumer as ^ + {`RdKafkaTPtr'} -> `RdKafkaRespErrT' cIntToEnum #} + +-- rd_kafka_assign +{#fun unsafe rd_kafka_assign as ^ + {`RdKafkaTPtr', `RdKafkaTopicPartitionListTPtr'} + -> `RdKafkaRespErrT' cIntToEnum #} + +{#fun unsafe rd_kafka_assignment as ^ + {`RdKafkaTPtr', castPtr `Ptr (Ptr RdKafkaTopicPartitionListT)'} + -> `RdKafkaRespErrT' cIntToEnum #} + +{#fun unsafe rd_kafka_commit as ^ + {`RdKafkaTPtr', `RdKafkaTopicPartitionListTPtr', `Int'} + -> `RdKafkaRespErrT' cIntToEnum #} + +{#fun unsafe rd_kafka_commit_message as ^ + {`RdKafkaTPtr', `RdKafkaMessageTPtr', `Int'} + -> `RdKafkaRespErrT' cIntToEnum #} + +{#fun unsafe rd_kafka_position as ^ + {`RdKafkaTPtr', `RdKafkaTopicPartitionListTPtr', `Int'} + -> `RdKafkaRespErrT' cIntToEnum #} +------------------------------------------------------------------------------------------------- +---- Groups +data RdKafkaGroupMemberInfoT = RdKafkaGroupMemberInfoT + { memberId'RdKafkaGroupMemberInfoT :: CString + , clientId'RdKafkaGroupMemberInfoT :: CString + , clientHost'RdKafkaGroupMemberInfoT :: CString + , memberMetadata'RdKafkaGroupMemberInfoT :: Word8Ptr + , memberMetadataSize'RdKafkaGroupMemberInfoT :: Int + , memberAssignment'RdKafkaGroupMemberInfoT :: Word8Ptr + , memberAssignmentSize'RdKafkaGroupMemberInfoT :: Int } + +instance Storable RdKafkaGroupMemberInfoT where + alignment _ = {#alignof rd_kafka_group_member_info#} + sizeOf _ = {#sizeof rd_kafka_group_member_info#} + peek p = RdKafkaGroupMemberInfoT + <$> liftM id ({#get rd_kafka_group_member_info->member_id #} p) + <*> liftM id ({#get rd_kafka_group_member_info->client_id #} p) + <*> liftM id ({#get rd_kafka_group_member_info->client_host #} p) + <*> liftM castPtr ({#get rd_kafka_group_member_info->member_metadata #} p) + <*> liftM fromIntegral ({#get rd_kafka_group_member_info->member_metadata_size #} p) + <*> liftM castPtr ({#get rd_kafka_group_member_info->member_assignment #} p) + <*> liftM fromIntegral ({#get rd_kafka_group_member_info->member_assignment_size #} p) + poke p x = do + {#set rd_kafka_group_member_info.member_id#} p (id $ memberId'RdKafkaGroupMemberInfoT x) + {#set rd_kafka_group_member_info.client_id#} p (id $ clientId'RdKafkaGroupMemberInfoT x) + {#set rd_kafka_group_member_info.client_host#} p (id $ clientHost'RdKafkaGroupMemberInfoT x) + {#set rd_kafka_group_member_info.member_metadata#} p (castPtr $ memberMetadata'RdKafkaGroupMemberInfoT x) + {#set rd_kafka_group_member_info.member_metadata_size#} p (fromIntegral $ memberMetadataSize'RdKafkaGroupMemberInfoT x) + {#set rd_kafka_group_member_info.member_assignment#} p (castPtr $ memberAssignment'RdKafkaGroupMemberInfoT x) + {#set rd_kafka_group_member_info.member_assignment_size#} p (fromIntegral $ memberAssignmentSize'RdKafkaGroupMemberInfoT x) + +{#pointer *rd_kafka_group_member_info as RdKafkaGroupMemberInfoTPtr foreign -> RdKafkaGroupMemberInfoT #} + +data RdKafkaGroupInfoT = RdKafkaGroupInfoT + { broker'RdKafkaGroupInfoT :: Ptr RdKafkaMetadataBrokerT + , group'RdKafkaGroupInfoT :: CString + , err'RdKafkaGroupInfoT :: RdKafkaRespErrT + , state'RdKafkaGroupInfoT :: CString + , protocolType'RdKafkaGroupInfoT :: CString + , protocol'RdKafkaGroupInfoT :: CString + , members'RdKafkaGroupInfoT :: Ptr RdKafkaGroupMemberInfoT + , memberCnt'RdKafkaGroupInfoT :: Int } + +instance Storable RdKafkaGroupInfoT where + alignment _ = {#alignof rd_kafka_group_info #} + sizeOf _ = {#sizeof rd_kafka_group_info #} + peek p = RdKafkaGroupInfoT + <$> liftM castPtr ({#get rd_kafka_group_info->broker #} p) + <*> liftM id ({#get rd_kafka_group_info->group #} p) + <*> liftM cIntToEnum ({#get rd_kafka_group_info->err #} p) + <*> liftM id ({#get rd_kafka_group_info->state #} p) + <*> liftM id ({#get rd_kafka_group_info->protocol_type #} p) + <*> liftM id ({#get rd_kafka_group_info->protocol #} p) + <*> liftM castPtr ({#get rd_kafka_group_info->members #} p) + <*> liftM fromIntegral ({#get rd_kafka_group_info->member_cnt #} p) + poke p x = do + {#set rd_kafka_group_info.broker#} p (castPtr $ broker'RdKafkaGroupInfoT x) + {#set rd_kafka_group_info.group#} p (id $ group'RdKafkaGroupInfoT x) + {#set rd_kafka_group_info.err#} p (enumToCInt $ err'RdKafkaGroupInfoT x) + {#set rd_kafka_group_info.state#} p (id $ state'RdKafkaGroupInfoT x) + {#set rd_kafka_group_info.protocol_type#} p (id $ protocolType'RdKafkaGroupInfoT x) + {#set rd_kafka_group_info.protocol#} p (id $ protocol'RdKafkaGroupInfoT x) + {#set rd_kafka_group_info.members#} p (castPtr $ members'RdKafkaGroupInfoT x) + {#set rd_kafka_group_info.member_cnt#} p (fromIntegral $ memberCnt'RdKafkaGroupInfoT x) + +{#pointer *rd_kafka_group_info as RdKafkaGroupInfoTPtr foreign -> RdKafkaGroupInfoT #} + +data RdKafkaGroupListT = RdKafkaGroupListT + { groups'RdKafkaGroupListT :: Ptr RdKafkaGroupInfoT + , groupCnt'RdKafkaGroupListT :: Int } + +instance Storable RdKafkaGroupListT where + alignment _ = {#alignof rd_kafka_group_list #} + sizeOf _ = {#sizeof rd_kafka_group_list #} + peek p = RdKafkaGroupListT + <$> liftM castPtr ({#get rd_kafka_group_list->groups #} p) + <*> liftM fromIntegral ({#get rd_kafka_group_list->group_cnt #} p) + poke p x = do + {#set rd_kafka_group_list.groups#} p (castPtr $ groups'RdKafkaGroupListT x) + {#set rd_kafka_group_list.group_cnt#} p (fromIntegral $ groupCnt'RdKafkaGroupListT x) + +{#pointer *rd_kafka_group_list as RdKafkaGroupListTPtr foreign -> RdKafkaGroupListT #} + +{#fun rd_kafka_list_groups as ^ + {`RdKafkaTPtr', `String', castPtr `Ptr (Ptr RdKafkaGroupListT)', `Int'} + -> `RdKafkaRespErrT' cIntToEnum #} + +foreign import ccall unsafe "rdkafka.h &rd_kafka_list_groups" + rdKafkaGroupListDestroy :: FunPtr (Ptr RdKafkaGroupListT -> IO ()) + +-- listRdKafkaGroups :: RdKafkaTPtr -> String -> Int -> IO (Either RdKafkaRespErrT RdKafkaGroupListTPtr) +-- listRdKafkaGroups k g t = alloca $ \lstDblPtr -> do +-- err <- rdKafkaListGroups k g lstDblPtr t +-- case err of +-- RdKafkaRespErrNoError -> do +-- lstPtr <- peek lstDblPtr +-- lst <- peek lstPtr +-- addForeignPtrFinalizer rdKafkaGroupListDestroy lstPtr +-- return $ Right lstPtr +-- e -> return $ Left e +------------------------------------------------------------------------------------------------- + -- rd_kafka_message -foreign import ccall unsafe "rdkafka.h &rd_kafka_message_destroy" - rdKafkaMessageDestroy :: FunPtr (Ptr RdKafkaMessageT -> IO ()) +foreign import ccall unsafe "rdkafka.h rd_kafka_message_destroy" + rdKafkaMessageDestroy :: Ptr RdKafkaMessageT -> IO () -- rd_kafka_conf {#fun unsafe rd_kafka_conf_new as ^ @@ -201,7 +634,7 @@ {`RdKafkaConfTPtr'} -> `RdKafkaConfTPtr' #} {#fun unsafe rd_kafka_conf_set as ^ - {`RdKafkaConfTPtr', `String', `String', id `CCharBufPointer', cIntConv `CSize'} + {`RdKafkaConfTPtr', `String', `String', id `CCharBufPointer', cIntConv `CSize'} -> `RdKafkaConfResT' cIntToEnum #} newRdKafkaConfT :: IO RdKafkaConfTPtr @@ -229,7 +662,7 @@ rdKafkaTopicConfDestroy :: FunPtr (Ptr RdKafkaTopicConfT -> IO ()) {#fun unsafe rd_kafka_topic_conf_set as ^ - {`RdKafkaTopicConfTPtr', `String', `String', id `CCharBufPointer', cIntConv `CSize'} + {`RdKafkaTopicConfTPtr', `String', `String', id `CCharBufPointer', cIntConv `CSize'} -> `RdKafkaConfResT' cIntToEnum #} newRdKafkaTopicConfT :: IO RdKafkaTopicConfTPtr @@ -243,14 +676,14 @@ -- rd_kafka {#fun unsafe rd_kafka_new as ^ - {enumToCInt `RdKafkaTypeT', `RdKafkaConfTPtr', id `CCharBufPointer', cIntConv `CSize'} + {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 = +newRdKafkaT kafkaType confPtr = allocaBytes nErrorBytes $ \charPtr -> do duper <- rdKafkaConfDup confPtr ret <- rdKafkaNew kafkaType duper charPtr (fromIntegral nErrorBytes) @@ -274,34 +707,34 @@ rdKafkaConsumeStart :: RdKafkaTopicTPtr -> Int -> Int64 -> IO (Maybe String) rdKafkaConsumeStart topicPtr partition offset = do i <- rdKafkaConsumeStartInternal topicPtr (fromIntegral partition) (fromIntegral offset) - case i of + 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 ^ +{#fun rd_kafka_consume as ^ {`RdKafkaTopicTPtr', cIntConv `CInt32T', `Int'} -> `RdKafkaMessageTPtr' #} -{#fun unsafe rd_kafka_consume_batch as ^ +{#fun 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 + case i of -1 -> kafkaErrnoString >>= return . Just _ -> return Nothing {#fun unsafe rd_kafka_offset_store as rdKafkaOffsetStore - {`RdKafkaTopicTPtr', cIntConv `CInt32T', cIntConv `CInt64T'} + {`RdKafkaTopicTPtr', cIntConv `CInt32T', cIntConv `CInt64T'} -> `RdKafkaRespErrT' cIntToEnum #} -- rd_kafka produce {#fun unsafe rd_kafka_produce as ^ - {`RdKafkaTopicTPtr', cIntConv `CInt32T', `Int', castPtr `Word8Ptr', + {`RdKafkaTopicTPtr', cIntConv `CInt32T', `Int', castPtr `Word8Ptr', cIntConv `CSize', castPtr `Word8Ptr', cIntConv `CSize', castPtr `Word8Ptr'} -> `Int' #} @@ -314,7 +747,7 @@ -- rd_kafka_metadata {#fun unsafe rd_kafka_metadata as ^ - {`RdKafkaTPtr', boolToCInt `Bool', `RdKafkaTopicTPtr', + {`RdKafkaTPtr', boolToCInt `Bool', `RdKafkaTopicTPtr', castMetadata `Ptr (Ptr RdKafkaMetadataT)', `Int'} -> `RdKafkaRespErrT' cIntToEnum #} @@ -332,6 +765,9 @@ -- rd_kafka_topic +{#fun unsafe rd_kafka_topic_name as ^ + {`RdKafkaTopicTPtr'} -> `String' #} + {#fun unsafe rd_kafka_topic_new as ^ {`RdKafkaTPtr', `String', `RdKafkaTopicConfTPtr'} -> `RdKafkaTopicTPtr' #} @@ -362,13 +798,13 @@ -- 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 diff --git a/src/Haskakafka/InternalSetup.hs b/src/Haskakafka/InternalSetup.hs --- a/src/Haskakafka/InternalSetup.hs +++ b/src/Haskakafka/InternalSetup.hs @@ -1,20 +1,20 @@ module Haskakafka.InternalSetup where -import Haskakafka.InternalTypes -import Haskakafka.InternalRdKafka -import Haskakafka.InternalRdKafkaEnum +import Haskakafka.InternalRdKafka +import Haskakafka.InternalRdKafkaEnum +import Haskakafka.InternalTypes -import Control.Exception -import Control.Monad -import Data.Map.Strict (Map) -import Foreign -import Foreign.C.String -import System.IO +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 +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 +-- 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 @@ -27,24 +27,24 @@ newKafkaPtr :: RdKafkaTypeT -> KafkaConf -> IO Kafka newKafkaPtr kafkaType c@(KafkaConf confPtr) = do - et <- newRdKafkaT kafkaType confPtr - case et of + 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 + 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 = +setLogLevel (Kafka kptr _) level = rdKafkaSetLogLevel kptr (fromEnum level) -- @@ -60,9 +60,9 @@ newKafkaConf :: IO KafkaConf newKafkaConf = newRdKafkaConfT >>= return . KafkaConf -kafkaConf :: ConfigOverrides -> IO (KafkaConf) -kafkaConf overrides = do - conf <- newKafkaConf +kafkaConf :: ConfigOverrides -> IO (KafkaConf) +kafkaConf overrides = do + conf <- newKafkaConf setAllKafkaConfValues conf overrides return conf @@ -73,9 +73,9 @@ return conf checkConfSetValue :: RdKafkaConfResT -> CCharBufPointer -> IO () -checkConfSetValue err charPtr = case err of +checkConfSetValue err charPtr = case err of RdKafkaConfOk -> return () - RdKafkaConfInvalid -> do + RdKafkaConfInvalid -> do str <- peekCString charPtr throw $ KafkaInvalidConfigurationValue str RdKafkaConfUnknown -> do @@ -83,7 +83,7 @@ throw $ KafkaUnknownConfigurationKey str setKafkaConfValue :: KafkaConf -> String -> String -> IO () -setKafkaConfValue (KafkaConf confPtr) key value = do +setKafkaConfValue (KafkaConf confPtr) key value = allocaBytes nErrorBytes $ \charPtr -> do err <- rdKafkaConfSet confPtr key value charPtr (fromIntegral nErrorBytes) checkConfSetValue err charPtr @@ -92,7 +92,7 @@ setAllKafkaConfValues conf overrides = forM_ overrides $ \(k, v) -> setKafkaConfValue conf k v setKafkaTopicConfValue :: KafkaTopicConf -> String -> String -> IO () -setKafkaTopicConfValue (KafkaTopicConf confPtr) key value = do +setKafkaTopicConfValue (KafkaTopicConf confPtr) key value = allocaBytes nErrorBytes $ \charPtr -> do err <- rdKafkaTopicConfSet confPtr key value charPtr (fromIntegral nErrorBytes) checkConfSetValue err charPtr @@ -113,15 +113,15 @@ 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 -> IO (Map String String) dumpConfFromKafka (Kafka _ cfg) = dumpKafkaConf cfg --- | Returns a map of the current topic configuration +-- | 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) = +dumpKafkaTopicConf (KafkaTopicConf kptr) = parseDump (\sizeptr -> rdKafkaTopicConfDump kptr sizeptr) dumpKafkaConf :: KafkaConf -> IO (Map String String) @@ -130,7 +130,7 @@ parseDump :: (CSizePtr -> IO (Ptr CString)) -> IO (Map String String) parseDump cstr = alloca $ \sizeptr -> do - strPtr <- cstr sizeptr + strPtr <- cstr sizeptr size <- peek sizeptr keysAndValues <- mapM (\i -> peekCString =<< peekElemOff strPtr i) [0..((fromIntegral size) - 1)] diff --git a/src/Haskakafka/InternalShared.hs b/src/Haskakafka/InternalShared.hs new file mode 100644 --- /dev/null +++ b/src/Haskakafka/InternalShared.hs @@ -0,0 +1,68 @@ +module Haskakafka.InternalShared +where + +import Control.Exception +import Control.Monad +import qualified Data.ByteString as BS +import qualified Data.ByteString.Internal as BSI +import Foreign +import Foreign.C.Error +import Haskakafka.InternalRdKafka +import Haskakafka.InternalRdKafkaEnum +import Haskakafka.InternalTypes + +word8PtrToBS :: Int -> Word8Ptr -> IO BS.ByteString +word8PtrToBS len ptr = BSI.create len $ \bsptr -> + BSI.memcpy bsptr ptr len + +fromMessagePtr :: RdKafkaMessageTPtr -> IO (Either KafkaError KafkaMessage) +fromMessagePtr ptr = + withForeignPtr ptr $ \realPtr -> + if realPtr == nullPtr then liftM (Left . kafkaRespErr) getErrno + else do + s <- peek realPtr + msg <- if err'RdKafkaMessageT s /= RdKafkaRespErrNoError + then return $ Left . KafkaResponseError $ err'RdKafkaMessageT s + else Right <$> fromMessageStorable s + rdKafkaMessageDestroy realPtr + return msg + +fromMessageStorable :: RdKafkaMessageT -> IO KafkaMessage +fromMessageStorable s = do + payload <- word8PtrToBS (len'RdKafkaMessageT s) (payload'RdKafkaMessageT s) + topic <- newForeignPtr_ (topic'RdKafkaMessageT s) >>= rdKafkaTopicName + + key <- if key'RdKafkaMessageT s == nullPtr + then return Nothing + else liftM Just $ word8PtrToBS (keyLen'RdKafkaMessageT s) (key'RdKafkaMessageT s) + + return $ KafkaMessage + topic + (partition'RdKafkaMessageT s) + (offset'RdKafkaMessageT s) + payload + key + +kafkaRespErr :: Errno -> KafkaError +kafkaRespErr (Errno num) = KafkaResponseError $ rdKafkaErrno2err (fromIntegral num) +{-# INLINE kafkaRespErr #-} + +throwOnError :: IO (Maybe String) -> IO () +throwOnError action = do + m <- action + case m of + Just e -> throw $ KafkaError e + Nothing -> return () + +hasError :: KafkaError -> Bool +hasError err = case err of + KafkaResponseError RdKafkaRespErrNoError -> False + _ -> True +{-# INLINE hasError #-} + +kafkaErrorToEither :: KafkaError -> Either KafkaError () +kafkaErrorToEither err = case err of + KafkaResponseError RdKafkaRespErrNoError -> Right () + _ -> Left err +{-# INLINE kafkaErrorToEither #-} + diff --git a/src/Haskakafka/InternalTypes.hs b/src/Haskakafka/InternalTypes.hs --- a/src/Haskakafka/InternalTypes.hs +++ b/src/Haskakafka/InternalTypes.hs @@ -2,41 +2,40 @@ module Haskakafka.InternalTypes where -import Control.Exception -import Data.Int -import Data.Typeable +import Control.Exception +import Data.Int +import Data.Typeable -import Haskakafka.InternalRdKafka -import Haskakafka.InternalRdKafkaEnum +import Haskakafka.InternalRdKafka +import Haskakafka.InternalRdKafkaEnum -import qualified Data.ByteString as BS +import qualified Data.ByteString as BS --- +-- -- Pointer wrappers -- -- | Kafka configuration object -data KafkaConf = KafkaConf RdKafkaConfTPtr +data KafkaConf = KafkaConf RdKafkaConfTPtr deriving (Show) -- | Kafka topic configuration object data KafkaTopicConf = KafkaTopicConf RdKafkaTopicConfTPtr -- | Main pointer to Kafka object, which contains our brokers -data Kafka = Kafka { kafkaPtr :: RdKafkaTPtr, _kafkaConf :: KafkaConf} +data Kafka = Kafka { kafkaPtr :: RdKafkaTPtr, _kafkaConf :: KafkaConf} deriving (Show) -- | 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 +data KafkaTopic = KafkaTopic + RdKafkaTopicTPtr + Kafka -- Kept around to prevent garbage collection KafkaTopicConf - -- -- Consumer -- -- | Starting locations for a consumer -data KafkaOffset = +data KafkaOffset = -- | Start reading from the beginning of the partition KafkaOffsetBeginning @@ -46,23 +45,26 @@ -- | Start reading from a specific location within the partition | KafkaOffset Int64 - -- | Start reading from the stored offset. See - -- + -- | Start reading from the stored offset. See + -- -- for offset store configuration. | KafkaOffsetStored + | KafkaOffsetInvalid + deriving (Eq, Show) -- | 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 +data KafkaMessage = + KafkaMessage { + messageTopic :: !String + -- | Kafka partition this message was received from + , messagePartition :: !Int -- | Offset within the 'messagePartition' Kafka partition - , messageOffset :: !Int64 + , messageOffset :: !Int64 -- | Contents of the message, as a 'ByteString' - , messagePayload :: !BS.ByteString + , messagePayload :: !BS.ByteString -- | Optional key of the message. 'Nothing' when the message -- was enqueued without a key - , messageKey :: Maybe BS.ByteString + , messageKey :: Maybe BS.ByteString } deriving (Eq, Show, Read, Typeable) @@ -71,74 +73,74 @@ -- -- | Represents messages /to be enqueued/ onto a Kafka broker (i.e. used for a producer) -data KafkaProduceMessage = +data KafkaProduceMessage = -- | A message without a key, assigned to 'KafkaSpecifiedPartition' or 'KafkaUnassignedPartition' - KafkaProduceMessage + KafkaProduceMessage {-# UNPACK #-} !BS.ByteString -- message payload -- | A message with a key, assigned to a partition based on the key - | KafkaProduceKeyedMessage + | 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 +data KafkaProducePartition = + -- | A specific partition in the topic KafkaSpecifiedPartition {-# UNPACK #-} !Int -- the partition number of the topic -- | A random partition within the topic - | KafkaUnassignedPartition + | KafkaUnassignedPartition -- -- Metadata -- -- | Metadata for all Kafka brokers data KafkaMetadata = KafkaMetadata - { + { -- | Broker metadata - brokers :: [KafkaBrokerMetadata] + brokers :: [KafkaBrokerMetadata] -- | topic metadata - , topics :: [Either KafkaError KafkaTopicMetadata] - } + , topics :: [Either KafkaError KafkaTopicMetadata] + } deriving (Eq, Show, Typeable) -- | Metadata for a specific Kafka broker data KafkaBrokerMetadata = KafkaBrokerMetadata - { + { -- | broker identifier - brokerId :: Int + brokerId :: Int -- | hostname for the broker - , brokerHost :: String + , brokerHost :: String -- | port for the broker - , brokerPort :: Int - } + , brokerPort :: Int + } deriving (Eq, Show, Typeable) -- | Metadata for a specific topic data KafkaTopicMetadata = KafkaTopicMetadata - { + { -- | name of the topic - topicName :: String + topicName :: String -- | partition metadata - , topicPartitions :: [Either KafkaError KafkaPartitionMetadata] + , topicPartitions :: [Either KafkaError KafkaPartitionMetadata] } deriving (Eq, Show, Typeable) -- | Metadata for a specific partition data KafkaPartitionMetadata = KafkaPartitionMetadata - { + { -- | identifier for the partition - partitionId :: Int + partitionId :: Int -- | broker leading this partition - , partitionLeader :: Int + , partitionLeader :: Int -- | replicas of the leader - , partitionReplicas :: [Int] + , partitionReplicas :: [Int] -- | In-sync replica set, see - , partitionIsrs :: [Int] - } + , partitionIsrs :: [Int] + } deriving (Eq, Show, Typeable) -- @@ -146,7 +148,7 @@ -- -- | Log levels for the RdKafkaLibrary used in 'setKafkaLogLevel' -data KafkaLogLevel = +data KafkaLogLevel = KafkaLogEmerg | KafkaLogAlert | KafkaLogCrit | KafkaLogErr | KafkaLogWarning | KafkaLogNotice | KafkaLogInfo | KafkaLogDebug @@ -171,7 +173,7 @@ fromEnum KafkaLogDebug = 7 -- | Any Kafka errors -data KafkaError = +data KafkaError = KafkaError String | KafkaInvalidReturnValue | KafkaBadSpecification String diff --git a/tests/TestMain.hs b/tests/TestMain.hs --- a/tests/TestMain.hs +++ b/tests/TestMain.hs @@ -1,4 +1,4 @@ -{-# LANGUAGE ScopedTypeVariables #-} +{-# LANGUAGE ScopedTypeVariables #-} module Main (main) where import Haskakafka import Haskakafka.InternalSetup @@ -26,7 +26,7 @@ cb b t sampleProduceMessages :: [KafkaProduceMessage] -sampleProduceMessages = +sampleProduceMessages = [ (KafkaProduceMessage $ C8.pack "hello") , (KafkaProduceKeyedMessage (C8.pack "key") (C8.pack "value")) , (KafkaProduceMessage $ C8.pack "goodbye") @@ -42,7 +42,7 @@ (Just pkey) `shouldBe` (messageKey m) primeEOF :: KafkaTopic -> IO () -primeEOF kt = consumeMessage kt 0 1000 >> return () +primeEOF kt = consumeMessage kt 0 kafkaDelay >> return () testmain :: IO () testmain = hspec $ do @@ -52,12 +52,12 @@ describe "Kafka Configuration" $ do it "should allow dumping" $ do - kConf <- newKafkaConf + kConf <- newKafkaConf kvs <- dumpKafkaConf kConf (Map.size kvs) `shouldSatisfy` (>0) it "should change when set is called" $ do - kConf <- newKafkaConf + kConf <- newKafkaConf setKafkaConfValue kConf "socket.timeout.ms" "50000" kvs <- dumpKafkaConf kConf (kvs Map.! "socket.timeout.ms") `shouldBe` "50000" @@ -71,7 +71,7 @@ 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 @@ -97,35 +97,37 @@ describe "Logging" $ do it "should allow setting of log level" $ getAddressTopic $ \a t -> do withKafkaConsumer [] [] a t 0 KafkaOffsetEnd $ \kafka _ -> do - setLogLevel kafka KafkaLogDebug + setLogLevel kafka KafkaLogDebug describe "Consume and produce cycle" $ do + it "should produce a single message" $ getAddressTopic $ \a t -> do + let message = KafkaProduceMessage (C8.pack "Hey, first test message!") + perr <- withKafkaProducer [] [] a t $ \_ producerTopic -> do + produceMessage producerTopic (KafkaSpecifiedPartition 0) message + perr `shouldBe`Nothing + 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 primeEOF topic - perr <- withKafkaProducer [] [] a t $ \_ producerTopic -> do + perr <- withKafkaProducer [] [] a t $ \_ producerTopic -> do produceMessage producerTopic (KafkaSpecifiedPartition 0) message perr `shouldBe` Nothing - + et <- consumeMessage topic 0 kafkaDelay - case et of + 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") + it "should be able to produce a keyed message" $ + getAddressTopic $ \a t -> do + let message = KafkaProduceKeyedMessage + (C8.pack "key") + (C8.pack "monkey around") - withKafkaConsumer [] [] a t 0 KafkaOffsetEnd $ \_ topic -> do - primeEOF topic - perr <- withKafkaProducer [] [] a t $ \_ producerTopic -> do + 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 + perr `shouldBe` Nothing it "should be able to batch produce messages" $ getAddressTopic $ \a t -> do withKafkaConsumer [] [] a t 0 KafkaOffsetEnd $ \_ topic -> do @@ -136,7 +138,7 @@ ets <- mapM (\_ -> consumeMessage topic 0 kafkaDelay) ([1..3] :: [Integer]) - forM_ (zip sampleProduceMessages ets) $ \(pm, et) -> + forM_ (zip sampleProduceMessages ets) $ \(pm, et) -> case (pm, et) of (_, Left err) -> error $ show err (pmessage, Right cm) -> pmessage `shouldBeProduceConsume` cm @@ -149,28 +151,45 @@ errs `shouldBe` [] et <- consumeMessageBatch topic 0 (5000) 3 - case et of + 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 for https://github.com/cosbynator/haskakafka/issues/12 + it "should not fail on batch consume when no messages are available #12" $ getAddressTopic $ \a t -> do + withKafkaConsumer [] [] a t 0 KafkaOffsetEnd $ \_ topic -> do + primeEOF topic + et <- consumeMessageBatch topic 0 (5000) 3 + case et of + (Left err) -> error $ show err + (Right oms) -> do + (length oms) `shouldBe` 0 + + it "should return EOF on batch consume if necessary" $ getAddressTopic $ \a t -> do + withKafkaConsumer [] [] a t 0 KafkaOffsetEnd $ \_ topic -> do + et <- consumeMessageBatch topic 0 (5000) 10 + case et of + (Left err) -> print err + (Right _oms) -> error "should return EOF" + -- Test setup (error on no Kafka) checkForKafka :: IO (Bool) checkForKafka = do a <- brokerAddress me <- fetchBrokerMetadata [] a 1000 - return $ case me of + return $ case me of (Left _) -> False (Right _) -> True -main :: IO () -main = do +main :: IO () +main = do a <- brokerAddress - hasKafka <- checkForKafka + hasKafka <- checkForKafka if hasKafka then testmain else error $ "\n\n\ \*******************************************************************************\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\ + \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"