haskakafka 1.0.0 → 1.1.0
raw patch · 14 files changed
+1365/−342 lines, 14 filesdep +cmdargsdep +pretty-showdep ~basenew-component:exe:basicnew-component:exe:simple
Dependencies added: cmdargs, pretty-show
Dependency ranges changed: base
Files
- example/Basic.hs +158/−0
- example/Simple.hs +75/−0
- haskakafka.cabal +34/−8
- src/Haskakafka.hs +89/−117
- src/Haskakafka/Consumer.hs +167/−0
- src/Haskakafka/Consumer/Internal/Convert.hs +71/−0
- src/Haskakafka/Consumer/Internal/Types.hs +38/−0
- src/Haskakafka/ConsumerExample.hs +62/−0
- src/Haskakafka/Example.hs +0/−71
- src/Haskakafka/InternalRdKafka.chs +477/−41
- src/Haskakafka/InternalSetup.hs +27/−27
- src/Haskakafka/InternalShared.hs +68/−0
- src/Haskakafka/InternalTypes.hs +51/−49
- tests/TestMain.hs +48/−29
+ example/Basic.hs view
@@ -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 "<brokers>"+ &= help "Comma separated list in format <hostname>:<port>,<hostname>:<port>"+ &= explicit+ &= name "brokers"+ &= name "b"+ , topic'CArgs = def+ &= typ "<topic>"+ &= 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 "<num>"+ &= 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
+ example/Simple.hs view
@@ -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)
haskakafka.cabal view
@@ -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
src/Haskakafka.hs view
@@ -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 <https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md>@@ -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 ()-
+ src/Haskakafka/Consumer.hs view
@@ -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)
+ src/Haskakafka/Consumer/Internal/Convert.hs view
@@ -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+
+ src/Haskakafka/Consumer/Internal/Types.hs view
@@ -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)+
+ src/Haskakafka/ConsumerExample.hs view
@@ -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)++
− src/Haskakafka/Example.hs
@@ -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)
src/Haskakafka/InternalRdKafka.chs view
@@ -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
src/Haskakafka/InternalSetup.hs view
@@ -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)]
@@ -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 #-}+
src/Haskakafka/InternalTypes.hs view
@@ -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 - -- <https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md librdkafka's documentation> + -- | Start reading from the stored offset. See+ -- <https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md librdkafka's documentation> -- 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 <http://kafka.apache.org/documentation.html>- , 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
tests/TestMain.hs view
@@ -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"