packages feed

haskakafka 0.5.0 → 1.0.0

raw patch · 10 files changed

+893/−697 lines, 10 files

Files

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