diff --git a/README.md b/README.md --- a/README.md +++ b/README.md @@ -1,33 +1,34 @@ # hw-kafka-client + [![CircleCI](https://circleci.com/gh/haskell-works/hw-kafka-client.svg?style=svg&circle-token=5f3ada2650dd600bc0fd4787143024867b2afc4e)](https://circleci.com/gh/haskell-works/hw-kafka-client) Kafka bindings for Haskell backed by the [librdkafka C module](https://github.com/edenhill/librdkafka). -## Credits -This project is inspired by [Haskakafka](https://github.com/cosbynator/haskakafka) -which unfortunately doesn't seem to be actively maintained. - ## Ecosystem -HaskellWorks Kafka ecosystem is described here: https://github.com/haskell-works/hw-kafka -# Consumer +HaskellWorks Kafka ecosystem is described here: + +## Consumer + High level consumers are supported by `librdkafka` starting from version 0.9. High-level consumers provide an abstraction for consuming messages from multiple partitions and topics. They are also address scalability (up to a number of partitions) by providing automatic rebalancing functionality. When a new consumer joins a consumer group the set of consumers attempt to "rebalance" the load to assign partitions to each consumer. -### Example: +### Consumer example +See [Running integration tests locally](#running-integration-tests-locally) to learn how to configure a local environment. + ```bash -$ stack build --flag hw-kafka-client:examples +cabal build --flag examples ``` or ```bash -$ stack build --exec kafka-client-example --flag hw-kafka-client:examples +cabal run kafka-client-example --flag examples ``` A working consumer example can be found here: [ConsumerExample.hs](example/ConsumerExample.hs)
@@ -75,7 +76,7 @@ return $ Right () ``` -# Producer +## Producer `kafka-client` producer supports sending messages to multiple topics. Target topic name is a part of each message that is to be sent by `produceMessage`. @@ -117,9 +118,9 @@ In the example above when the producer cannot deliver the message to Kafka, the error will be printed (and the message will be dropped). -### Example +### Producer example -```Haskell +```haskell {-# LANGUAGE OverloadedStrings #-} import Control.Exception (bracket) import Control.Monad (forM_) @@ -166,6 +167,7 @@ ``` ### Synchronous sending of messages + Because of the asynchronous nature of librdkafka, there is no API to provide synchronous production of messages. It is, however, possible to combine the delivery reports feature with that of callbacks. This can be done using the @@ -215,44 +217,27 @@ librdkafka times out). You should make sure that your configuration reflects the behavior you want out of this functionality. -# Installation +## Running integration tests locally -## Installing librdkafka +[shell.nix](./shell.nix) can be used to provide a working environment that is enough to build and test `hw-kafka-client`. -Although `librdkafka` is available on many platforms, most of -the distribution packages are too old to support `kafka-client`. -As such, we suggest you install from the source: +To be able to run tests locally, `$KAFKA_TEST_BROKER` environment variable is expected to be set (use [shell.nix](./shell.nix) or export manually). -```bash - git clone https://github.com/edenhill/librdkafka - cd librdkafka - ./configure - make && make install -``` +`$KAFKA_TEST_BROKER` should contain an IP address of an accessible Kafka broker that will be used to run integration tests against. -Sometimes it is helpful to specify openssl includes explicitly: +With [Docker Compose](./docker-compose.yml) this variable is used to configure Kafka broker to listen on this address: ``` - LDFLAGS=-L/usr/local/opt/openssl/lib CPPFLAGS=-I/usr/local/opt/openssl/include ./configure +$ docker-compose up ``` -If you are using Stack with Nix, don't forget to declare `rdkafka` as extra package: +After that, tests can be run as usual: -```yaml -# stack.yaml -nix: - enable: true - packages: - - rdkafka ``` - -## Installing Kafka - -The full Kafka guide is at http://kafka.apache.org/documentation.html#quickstart +$ cabal test --test-show-details=direct +``` -Alternatively `docker-compose` can be used to run Kafka locally inside a Docker container. -To run Kafka inside Docker: +## Credits -```bash -$ docker-compose up -``` +This project is inspired by [Haskakafka](https://github.com/cosbynator/haskakafka) +which unfortunately doesn't seem to be actively maintained. diff --git a/hw-kafka-client.cabal b/hw-kafka-client.cabal --- a/hw-kafka-client.cabal +++ b/hw-kafka-client.cabal @@ -1,7 +1,7 @@ cabal-version: 2.2 name: hw-kafka-client -version: 3.1.0 +version: 3.1.1 synopsis: Kafka bindings for Haskell description: Apache Kafka bindings backed by the librdkafka C library. . @@ -87,6 +87,7 @@ test-suite integration-tests type: exitcode-stdio-1.0 + ghc-options: -threaded -rtsopts -with-rtsopts=-N main-is: Spec.hs hs-source-dirs: tests-it ghc-options: -Wall -threaded @@ -108,6 +109,7 @@ test-suite tests type: exitcode-stdio-1.0 + ghc-options: -threaded -rtsopts -with-rtsopts=-N main-is: Spec.hs hs-source-dirs: tests ghc-options: -Wall -threaded diff --git a/src/Kafka/Callbacks.hs b/src/Kafka/Callbacks.hs --- a/src/Kafka/Callbacks.hs +++ b/src/Kafka/Callbacks.hs @@ -7,18 +7,48 @@ import Kafka.Internal.RdKafka (rdKafkaConfSetErrorCb, rdKafkaConfSetLogCb, rdKafkaConfSetStatsCb) import Kafka.Internal.Setup (HasKafkaConf(..), getRdKafkaConf) -import Kafka.Types (KafkaError(..)) +import Kafka.Types (KafkaError(..), KafkaLogLevel(..)) +-- | Add a callback for errors. +-- +-- ==== __Examples__ +-- +-- Basic usage: +-- +-- > setCallback (errorCallback myErrorCallback) +-- > +-- > myErrorCallback :: KafkaError -> String -> IO () +-- > myErrorCallback kafkaError message = print $ show kafkaError <> "|" <> message errorCallback :: HasKafkaConf k => (KafkaError -> String -> IO ()) -> k -> IO () errorCallback callback k = let realCb _ err = callback (KafkaResponseError err) in rdKafkaConfSetErrorCb (getRdKafkaConf k) realCb -logCallback :: HasKafkaConf k => (Int -> String -> String -> IO ()) -> k -> IO () +-- | Add a callback for logs. +-- +-- ==== __Examples__ +-- +-- Basic usage: +-- +-- > setCallback (logCallback myLogCallback) +-- > +-- > myLogCallback :: KafkaLogLevel -> String -> String -> IO () +-- > myLogCallback level facility message = print $ show level <> "|" <> facility <> "|" <> message +logCallback :: HasKafkaConf k => (KafkaLogLevel -> String -> String -> IO ()) -> k -> IO () logCallback callback k = - let realCb _ = callback + let realCb _ = callback . toEnum in rdKafkaConfSetLogCb (getRdKafkaConf k) realCb +-- | Add a callback for stats. +-- +-- ==== __Examples__ +-- +-- Basic usage: +-- +-- > setCallback (statsCallback myStatsCallback) +-- > +-- > myStatsCallback :: String -> IO () +-- > myStatsCallback stats = print $ show stats statsCallback :: HasKafkaConf k => (String -> IO ()) -> k -> IO () statsCallback callback k = let realCb _ = callback diff --git a/src/Kafka/Consumer.hs b/src/Kafka/Consumer.hs --- a/src/Kafka/Consumer.hs +++ b/src/Kafka/Consumer.hs @@ -1,8 +1,58 @@ {-# LANGUAGE LambdaCase #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE TupleSections #-} + +----------------------------------------------------------------------------- +-- | +-- Module to consume messages from Kafka topics. +-- +-- Here's an example of code to consume messages from a topic: +-- +-- @ +-- import Control.Exception (bracket) +-- import Data.Monoid ((<>)) +-- import Kafka +-- import Kafka.Consumer +-- +-- -- Global consumer properties +-- consumerProps :: 'ConsumerProperties' +-- consumerProps = 'brokersList' ['BrokerAddress' "localhost:9092"] +-- <> 'groupId' ('ConsumerGroupId' "consumer_example_group") +-- <> 'noAutoCommit' +-- <> 'logLevel' 'KafkaLogInfo' +-- +-- -- Subscription to topics +-- consumerSub :: 'Subscription' +-- consumerSub = 'topics' ['TopicName' "kafka-client-example-topic"] +-- <> 'offsetReset' 'Earliest' +-- +-- -- Running an example +-- runConsumerExample :: IO () +-- runConsumerExample = do +-- res <- bracket mkConsumer clConsumer runHandler +-- print res +-- where +-- mkConsumer = 'newConsumer' consumerProps consumerSub +-- clConsumer (Left err) = return (Left err) +-- clConsumer (Right kc) = (maybe (Right ()) Left) \<$\> 'closeConsumer' kc +-- runHandler (Left err) = return (Left err) +-- runHandler (Right kc) = processMessages kc +-- +-- -- Example polling 10 times before stopping +-- processMessages :: 'KafkaConsumer' -> IO (Either 'KafkaError' ()) +-- processMessages kafka = do +-- mapM_ (\_ -> do +-- msg1 <- 'pollMessage' kafka ('Timeout' 1000) +-- putStrLn $ "Message: " <> show msg1 +-- err <- 'commitAllOffsets' 'OffsetCommit' kafka +-- putStrLn $ "Offsets: " <> maybe "Committed." show err +-- ) [0 .. 10] +-- return $ Right () +-- @ +----------------------------------------------------------------------------- module Kafka.Consumer -( module X +( KafkaConsumer +, module X , runConsumer , newConsumer , assign, assignment, subscription @@ -14,7 +64,6 @@ , storeOffsets, storeOffsetMessage , closeConsumer -- ReExport Types -, KafkaConsumer , RdKafkaRespErrT (..) ) where @@ -48,7 +97,7 @@ -- | Runs high-level kafka consumer. -- A callback provided is expected to call 'pollMessage' when convenient. -{-# DEPRECATED runConsumer "Use newConsumer/closeConsumer instead" #-} +{-# DEPRECATED runConsumer "Use 'newConsumer'/'closeConsumer' instead" #-} runConsumer :: ConsumerProperties -> Subscription -> (KafkaConsumer -> IO (Either KafkaError a)) -- ^ A callback function to poll and handle messages @@ -64,6 +113,7 @@ runHandler (Left err) = return (Left err) runHandler (Right kc) = f kc +-- | Create a `KafkaConsumer`. This consumer must be correctly released using 'closeConsumer'. newConsumer :: MonadIO m => ConsumerProperties -> Subscription @@ -94,6 +144,7 @@ runConsumerLoop kafka (Just $ Timeout 100)) >> return (Right kafka) Just err -> closeConsumer kafka >> return (Left err) +-- | Polls a single message pollMessage :: MonadIO m => KafkaConsumer -> Timeout -- ^ the timeout, in milliseconds @@ -104,7 +155,7 @@ Nothing -> rdKafkaConsumerPoll (getRdKafka c) ms >>= fromMessagePtr Just q -> rdKafkaConsumeQueue q (fromIntegral ms) >>= fromMessagePtr --- | Polls up to BatchSize messages. +-- | Polls up to 'BatchSize' messages. -- Unlike 'pollMessage' this function does not return usual "timeout" errors. -- An empty batch is returned when there are no messages available. -- @@ -205,6 +256,7 @@ mapM_ (\(TopicName topicName, PartitionId partitionId) -> rdKafkaTopicPartitionListAdd pl (Text.unpack topicName) partitionId) ps KafkaResponseError <$> rdKafkaResumePartitions k pl +-- | Seek a particular offset for each provided 'TopicPartition' seek :: MonadIO m => KafkaConsumer -> Timeout -> [TopicPartition] -> m (Maybe KafkaError) seek (KafkaConsumer (Kafka k) _) (Timeout timeout) tps = liftIO $ either Just (const Nothing) <$> seekAll @@ -244,7 +296,7 @@ -- -- Events will cause application provided callbacks to be called. -- --- The \p timeout_ms argument specifies the maximum amount of time +-- The 'Timeout' argument specifies the maximum amount of time -- (in milliseconds) that the call will block waiting for events. -- -- This function is called on each 'pollMessage' and, if runtime allows @@ -260,6 +312,8 @@ void . withCallbackPollEnabled k $ pollConsumerEvents' k timeout -- | Closes the consumer. +-- +-- See 'newConsumer' closeConsumer :: MonadIO m => KafkaConsumer -> m (Maybe KafkaError) closeConsumer (KafkaConsumer (Kafka k) (KafkaConf _ qr statusVar)) = liftIO $ -- because closing the consumer will raise callbacks, @@ -280,7 +334,7 @@ -- | Subscribes to a given list of topics. -- --- Wildcard (regex) topics are supported by the librdkafka assignor: +-- 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. diff --git a/src/Kafka/Consumer/Subscription.hs b/src/Kafka/Consumer/Subscription.hs --- a/src/Kafka/Consumer/Subscription.hs +++ b/src/Kafka/Consumer/Subscription.hs @@ -17,6 +17,18 @@ import Kafka.Consumer.Types (OffsetReset (..)) import Kafka.Types (TopicName (..)) +-- | A consumer subscription to a topic. +-- +-- ==== __Examples__ +-- +-- Typically you don't call the constructor directly, but combine settings: +-- +-- @ +-- consumerSub :: 'Subscription' +-- consumerSub = 'topics' ['TopicName' "kafka-client-example-topic"] +-- <> 'offsetReset' 'Earliest' +-- <> 'extraSubscriptionProps' (fromList [("prop1", "value 1"), ("prop2", "value 2")]) +-- @ data Subscription = Subscription (Set TopicName) (Map Text Text) instance Sem.Semigroup Subscription where @@ -32,9 +44,11 @@ mappend = (Sem.<>) {-# INLINE mappend #-} +-- | Build a subscription by giving the list of topic names only topics :: [TopicName] -> Subscription topics ts = Subscription (Set.fromList ts) M.empty +-- | Build a subscription by giving the offset reset parameter only offsetReset :: OffsetReset -> Subscription offsetReset o = let o' = case o of @@ -42,5 +56,6 @@ Latest -> "latest" in Subscription Set.empty (M.fromList [("auto.offset.reset", o')]) +-- | Build a subscription by giving extra properties only extraSubscriptionProps :: Map Text Text -> Subscription extraSubscriptionProps = Subscription Set.empty diff --git a/src/Kafka/Consumer/Types.hs b/src/Kafka/Consumer/Types.hs --- a/src/Kafka/Consumer/Types.hs +++ b/src/Kafka/Consumer/Types.hs @@ -30,7 +30,7 @@ import Data.Bifoldable (Bifoldable (..)) import Data.Bifunctor (Bifunctor (..)) -import Data.Bitraversable (Bitraversable (..), bimapM, bisequenceA) +import Data.Bitraversable (Bitraversable (..), bimapM, bisequence) import Data.Int (Int64) import Data.Text (Text) import Data.Typeable (Typeable) @@ -38,6 +38,9 @@ import Kafka.Internal.Setup (HasKafka (..), HasKafkaConf (..), Kafka (..), KafkaConf (..)) import Kafka.Types (Millis (..), PartitionId (..), TopicName (..)) +-- | The main type for Kafka consumption, used e.g. to poll and commit messages. +-- +-- Its constructor is intentionally not exposed, instead, one should used 'newConsumer' to acquire such a value. data KafkaConsumer = KafkaConsumer { kcKafkaPtr :: !Kafka , kcKafkaConf :: !KafkaConf @@ -51,8 +54,17 @@ getKafkaConf = kcKafkaConf {-# INLINE getKafkaConf #-} +-- | Consumer group ID. Different consumers with the same consumer group ID will get assigned different partitions of each subscribed topic. +-- +-- See newtype ConsumerGroupId = ConsumerGroupId { unConsumerGroupId :: Text } deriving (Show, Ord, Eq, Generic) + +-- | A message offset in a partition newtype Offset = Offset { unOffset :: Int64 } deriving (Show, Eq, Ord, Read, Generic) + +-- | Where to reset the offset when there is no initial offset in Kafka +-- +-- See data OffsetReset = Earliest | Latest deriving (Show, Eq, Generic) -- | A set of events which happen during the rebalancing process @@ -67,6 +79,7 @@ | RebalanceRevoke [(TopicName, PartitionId)] deriving (Eq, Show, Generic) +-- | The partition offset data PartitionOffset = PartitionOffsetBeginning | PartitionOffsetEnd @@ -75,11 +88,13 @@ | PartitionOffsetInvalid deriving (Eq, Show, Generic) +-- | Partitions subscribed by a consumer data SubscribedPartitions - = SubscribedPartitions [PartitionId] - | SubscribedPartitionsAll + = SubscribedPartitions [PartitionId] -- ^ Subscribe only to those partitions + | SubscribedPartitionsAll -- ^ Subscribe to all partitions deriving (Show, Eq, Generic) +-- | Consumer record timestamp data Timestamp = CreateTime !Millis | LogAppendTime !Millis @@ -119,8 +134,8 @@ , crPartition :: !PartitionId -- ^ Kafka partition this message was received from , crOffset :: !Offset -- ^ Offset within the 'crPartition' Kafka partition , crTimestamp :: !Timestamp -- ^ Message timestamp - , crKey :: !k - , crValue :: !v + , crKey :: !k -- ^ Message key + , crValue :: !v -- ^ Message value } deriving (Eq, Show, Read, Typeable, Generic) @@ -148,24 +163,27 @@ bitraverse f g r = (\k v -> bimap (const k) (const v) r) <$> f (crKey r) <*> g (crValue r) {-# INLINE bitraverse #-} +{-# DEPRECATED crMapKey "Isn't concern of this library. Use 'first'" #-} crMapKey :: (k -> k') -> ConsumerRecord k v -> ConsumerRecord k' v crMapKey = first {-# INLINE crMapKey #-} +{-# DEPRECATED crMapValue "Isn't concern of this library. Use 'second'" #-} crMapValue :: (v -> v') -> ConsumerRecord k v -> ConsumerRecord k v' crMapValue = second {-# INLINE crMapValue #-} +{-# DEPRECATED crMapKV "Isn't concern of this library. Use 'bimap'" #-} crMapKV :: (k -> k') -> (v -> v') -> ConsumerRecord k v -> ConsumerRecord k' v' crMapKV = bimap {-# INLINE crMapKV #-} -{-# DEPRECATED sequenceFirst "Isn't concern of this library. Use 'bitraverse id pure'" #-} +{-# DEPRECATED sequenceFirst "Isn't concern of this library. Use @'bitraverse' 'id' 'pure'@" #-} sequenceFirst :: (Bitraversable t, Applicative f) => t (f k) v -> f (t k v) sequenceFirst = bitraverse id pure {-# INLINE sequenceFirst #-} -{-# DEPRECATED traverseFirst "Isn't concern of this library. Use 'bitraverse f pure'" #-} +{-# DEPRECATED traverseFirst "Isn't concern of this library. Use @'bitraverse' f 'pure'@" #-} traverseFirst :: (Bitraversable t, Applicative f) => (k -> f k') -> t k v @@ -173,7 +191,7 @@ traverseFirst f = bitraverse f pure {-# INLINE traverseFirst #-} -{-# DEPRECATED traverseFirstM "Isn't concern of this library. Use 'bitraverse id pure <$> bitraverse f pure r'" #-} +{-# DEPRECATED traverseFirstM "Isn't concern of this library. Use @'bitraverse' 'id' 'pure' '<$>' 'bitraverse' f 'pure' r@" #-} traverseFirstM :: (Bitraversable t, Applicative f, Monad m) => (k -> m (f k')) -> t k v @@ -181,7 +199,7 @@ traverseFirstM f r = bitraverse id pure <$> bitraverse f pure r {-# INLINE traverseFirstM #-} -{-# DEPRECATED traverseM "Isn't concern of this library. Use 'sequenceA <$> traverse f r'" #-} +{-# DEPRECATED traverseM "Isn't concern of this library. Use @'sequenceA' '<$>' 'traverse' f r@" #-} traverseM :: (Traversable t, Applicative f, Monad m) => (v -> m (f v')) -> t v @@ -189,12 +207,12 @@ traverseM f r = sequenceA <$> traverse f r {-# INLINE traverseM #-} -{-# DEPRECATED bitraverseM "Isn't concern of this library. Use 'bisequenceA <$> bimapM f g r'" #-} +{-# DEPRECATED bitraverseM "Isn't concern of this library. Use @'bisequenceA' '<$>' 'bimapM' f g r@" #-} bitraverseM :: (Bitraversable t, Applicative f, Monad m) => (k -> m (f k')) -> (v -> m (f v')) -> t k v -> m (f (t k' v')) -bitraverseM f g r = bisequenceA <$> bimapM f g r +bitraverseM f g r = bisequence <$> bimapM f g r {-# INLINE bitraverseM #-} diff --git a/src/Kafka/Internal/RdKafka.chs b/src/Kafka/Internal/RdKafka.chs --- a/src/Kafka/Internal/RdKafka.chs +++ b/src/Kafka/Internal/RdKafka.chs @@ -8,11 +8,12 @@ import Control.Monad (liftM) import Data.Int (Int32, Int64) import Data.Word (Word8) +import Foreign.Concurrent (newForeignPtr) import Foreign.Marshal.Alloc (alloca, allocaBytes) import Foreign.Marshal.Array (peekArray, allocaArray) import Foreign.Storable (Storable(..)) import Foreign.Ptr (Ptr, FunPtr, castPtr, nullPtr) -import Foreign.ForeignPtr (FinalizerPtr, addForeignPtrFinalizer, withForeignPtr, newForeignPtr, newForeignPtr_) +import Foreign.ForeignPtr (FinalizerPtr, addForeignPtrFinalizer, newForeignPtr_, withForeignPtr) import Foreign.C.Error (Errno(..), getErrno) import Foreign.C.String (CString, newCString, withCAString, peekCAString, peekCAStringLen, peekCString) import Foreign.C.Types (CFile, CInt(..), CSize, CChar) @@ -267,12 +268,14 @@ {`Int'} -> `RdKafkaTopicPartitionListTPtr' #} foreign import ccall unsafe "rdkafka.h &rd_kafka_topic_partition_list_destroy" - rdKafkaTopicPartitionListDestroy :: FinalizerPtr RdKafkaTopicPartitionListT + rdKafkaTopicPartitionListDestroyF :: FinalizerPtr RdKafkaTopicPartitionListT +foreign import ccall unsafe "rdkafka.h rd_kafka_topic_partition_list_destroy" + rdKafkaTopicPartitionListDestroy :: Ptr RdKafkaTopicPartitionListT -> IO () newRdKafkaTopicPartitionListT :: Int -> IO RdKafkaTopicPartitionListTPtr newRdKafkaTopicPartitionListT size = do ret <- rdKafkaTopicPartitionListNew size - addForeignPtrFinalizer rdKafkaTopicPartitionListDestroy ret + addForeignPtrFinalizer rdKafkaTopicPartitionListDestroyF ret return ret {# fun rd_kafka_topic_partition_list_add as ^ @@ -287,7 +290,7 @@ copyRdKafkaTopicPartitionList :: RdKafkaTopicPartitionListTPtr -> IO RdKafkaTopicPartitionListTPtr copyRdKafkaTopicPartitionList pl = do cp <- rdKafkaTopicPartitionListCopy pl - addForeignPtrFinalizer rdKafkaTopicPartitionListDestroy cp + addForeignPtrFinalizer rdKafkaTopicPartitionListDestroyF cp return cp {# fun rd_kafka_topic_partition_list_set_offset as ^ @@ -560,7 +563,7 @@ rdKafkaConsumeBatchQueue qptr timeout batchSize = do allocaArray batchSize $ \pArr -> do rSize <- rdKafkaConsumeBatchQueue' qptr timeout pArr (fromIntegral batchSize) - peekArray (fromIntegral rSize) pArr >>= traverse newForeignPtr_ + peekArray (fromIntegral rSize) pArr >>= traverse (flip newForeignPtr (return ())) ------------------------------------------------------------------------------------------------- ---- High-level KafkaConsumer @@ -582,7 +585,7 @@ (err, sub) <- rdKafkaSubscription' k case err of RdKafkaRespErrNoError -> - Right <$> newForeignPtr rdKafkaTopicPartitionListDestroy sub + Right <$> newForeignPtr sub (rdKafkaTopicPartitionListDestroy sub) e -> return (Left e) {#fun rd_kafka_consumer_poll as ^ @@ -614,7 +617,7 @@ (err, ass) <- rdKafkaAssignment' k case err of RdKafkaRespErrNoError -> - Right <$> newForeignPtr rdKafkaTopicPartitionListDestroy ass + Right <$> newForeignPtr ass (rdKafkaTopicPartitionListDestroy ass) e -> return (Left e) {#fun rd_kafka_commit as ^ @@ -721,8 +724,11 @@ -> `RdKafkaRespErrT' cIntToEnum #} foreign import ccall "rdkafka.h &rd_kafka_group_list_destroy" - rdKafkaGroupListDestroy :: FinalizerPtr RdKafkaGroupListT + rdKafkaGroupListDestroyF :: FinalizerPtr RdKafkaGroupListT +foreign import ccall "rdkafka.h rd_kafka_group_list_destroy" + rdKafkaGroupListDestroy :: Ptr RdKafkaGroupListT -> IO () + rdKafkaListGroups :: RdKafkaTPtr -> Maybe String -> Int -> IO (Either RdKafkaRespErrT RdKafkaGroupListTPtr) rdKafkaListGroups k g t = case g of Nothing -> listGroups nullPtr @@ -731,7 +737,7 @@ listGroups grp = do (err, res) <- rdKafkaListGroups' k grp t case err of - RdKafkaRespErrNoError -> Right <$> newForeignPtr rdKafkaGroupListDestroy res + RdKafkaRespErrNoError -> Right <$> newForeignPtr res (rdKafkaGroupListDestroy res) e -> return $ Left e ------------------------------------------------------------------------------------------------- @@ -910,15 +916,15 @@ alloca- `Ptr RdKafkaMetadataT' peekPtr*, `Int'} -> `RdKafkaRespErrT' cIntToEnum #} -foreign import ccall unsafe "rdkafka.h &rd_kafka_metadata_destroy" - rdKafkaMetadataDestroy :: FinalizerPtr RdKafkaMetadataT +foreign import ccall unsafe "rdkafka.h rd_kafka_metadata_destroy" + rdKafkaMetadataDestroy :: Ptr RdKafkaMetadataT -> IO () rdKafkaMetadata :: RdKafkaTPtr -> Bool -> Maybe RdKafkaTopicTPtr -> Int -> IO (Either RdKafkaRespErrT RdKafkaMetadataTPtr) rdKafkaMetadata k allTopics mt timeout = do tptr <- maybe (newForeignPtr_ nullPtr) pure mt (err, res) <- rdKafkaMetadata' k allTopics tptr timeout case err of - RdKafkaRespErrNoError -> Right <$> newForeignPtr rdKafkaMetadataDestroy res + RdKafkaRespErrNoError -> Right <$> newForeignPtr res (rdKafkaMetadataDestroy res) e -> return (Left e) {#fun rd_kafka_poll as ^ diff --git a/src/Kafka/Producer.hs b/src/Kafka/Producer.hs --- a/src/Kafka/Producer.hs +++ b/src/Kafka/Producer.hs @@ -41,7 +41,7 @@ -- | Runs Kafka Producer. -- The callback provided is expected to call 'produceMessage' -- or/and 'produceMessageBatch' to send messages to Kafka. -{-# DEPRECATED runProducer "Use newProducer/closeProducer instead" #-} +{-# DEPRECATED runProducer "Use 'newProducer'/'closeProducer' instead" #-} runProducer :: ProducerProperties -> (KafkaProducer -> IO (Either KafkaError a)) -> IO (Either KafkaError a) diff --git a/src/Kafka/Producer/Callbacks.hs b/src/Kafka/Producer/Callbacks.hs --- a/src/Kafka/Producer/Callbacks.hs +++ b/src/Kafka/Producer/Callbacks.hs @@ -6,11 +6,12 @@ where import Control.Monad (void) +import Control.Exception (bracket) import Control.Concurrent (forkIO) import Foreign.C.Error (getErrno) import Foreign.Ptr (Ptr, nullPtr) import Foreign.Storable (Storable(peek)) -import Foreign.StablePtr (castPtrToStablePtr, deRefStablePtr) +import Foreign.StablePtr (castPtrToStablePtr, deRefStablePtr, freeStablePtr) import Kafka.Callbacks as X import Kafka.Consumer.Types (Offset(..)) import Kafka.Internal.RdKafka (RdKafkaMessageT(..), RdKafkaRespErrT(..), rdKafkaConfSetDrMsgCb) @@ -44,8 +45,8 @@ callback rep if cbPtr == nullPtr then pure () - else do - msgCb <- deRefStablePtr @(DeliveryReport -> IO ()) $ castPtrToStablePtr $ cbPtr + else bracket (pure $ castPtrToStablePtr cbPtr) freeStablePtr $ \stablePtr -> do + msgCb <- deRefStablePtr @(DeliveryReport -> IO ()) stablePtr -- Here we fork the callback since it might be a longer action and -- blocking here would block librdkafka from continuing its execution void . forkIO $ msgCb rep diff --git a/src/Kafka/Types.hs b/src/Kafka/Types.hs --- a/src/Kafka/Types.hs +++ b/src/Kafka/Types.hs @@ -29,16 +29,27 @@ import GHC.Generics (Generic) import Kafka.Internal.RdKafka (RdKafkaRespErrT, rdKafkaErr2name, rdKafkaErr2str) +-- | Kafka broker ID newtype BrokerId = BrokerId { unBrokerId :: Int } deriving (Show, Eq, Ord, Read, Generic) +-- | Topic partition ID newtype PartitionId = PartitionId { unPartitionId :: Int } deriving (Show, Eq, Read, Ord, Enum, Generic) + +-- | A number of milliseconds, used to represent durations and timestamps newtype Millis = Millis { unMillis :: Int64 } deriving (Show, Read, Eq, Ord, Num, Generic) + +-- | Client ID used by Kafka to better track requests +-- +-- See newtype ClientId = ClientId { unClientId :: Text } deriving (Show, Eq, Ord, Generic) + +-- | Batch size used for polling newtype BatchSize = BatchSize { unBatchSize :: Int } deriving (Show, Read, Eq, Ord, Num, Generic) +-- | Whether the topic is created by a user or by the system data TopicType = User -- ^ Normal topics that are created by user. - | System -- ^ Topics starting with "__" (__consumer_offsets, __confluent.support.metrics) are considered "system" topics + | System -- ^ Topics starting with a double underscore "\__" (@__consumer_offsets@, @__confluent.support.metrics@, etc.) are considered "system" topics deriving (Show, Read, Eq, Ord, Generic) -- | Topic name to be consumed @@ -51,6 +62,7 @@ TopicName { unTopicName :: Text } -- ^ a simple topic name or a regex if started with @^@ deriving (Show, Eq, Ord, Read, Generic) +-- | Deduce the type of a topic from its name, by checking if it starts with a double underscore "\__" topicType :: TopicName -> TopicType topicType (TopicName tn) = if "__" `isPrefixOf` tn then System else User @@ -68,8 +80,7 @@ KafkaLogNotice | KafkaLogInfo | KafkaLogDebug deriving (Show, Enum, Eq) --- --- | Any Kafka errors +-- | All possible Kafka errors data KafkaError = KafkaError Text | KafkaInvalidReturnValue @@ -85,6 +96,7 @@ "[" ++ rdKafkaErr2name err ++ "] " ++ rdKafkaErr2str err displayException err = show err +-- | Available /librdkafka/ debug contexts data KafkaDebug = DebugGeneric | DebugBroker @@ -100,6 +112,9 @@ | DebugAll deriving (Eq, Show, Typeable, Generic) +-- | Convert a 'KafkaDebug' into its /librdkafka/ string equivalent. +-- +-- This is used internally by the library but may be useful to some developers. kafkaDebugToText :: KafkaDebug -> Text kafkaDebugToText d = case d of DebugGeneric -> "generic" @@ -115,6 +130,9 @@ DebugFeature -> "feature" DebugAll -> "all" +-- | Compression codec used by a topic +-- +-- See data KafkaCompressionCodec = NoCompression | Gzip @@ -122,6 +140,9 @@ | Lz4 deriving (Eq, Show, Typeable, Generic) +-- | Convert a 'KafkaCompressionCodec' into its /librdkafka/ string equivalent. +-- +-- This is used internally by the library but may be useful to some developers. kafkaCompressionCodecToText :: KafkaCompressionCodec -> Text kafkaCompressionCodecToText c = case c of NoCompression -> "none" diff --git a/tests-it/Kafka/IntegrationSpec.hs b/tests-it/Kafka/IntegrationSpec.hs --- a/tests-it/Kafka/IntegrationSpec.hs +++ b/tests-it/Kafka/IntegrationSpec.hs @@ -5,22 +5,21 @@ module Kafka.IntegrationSpec where -import Control.Monad (forM, forM_) -import Control.Monad.Loops -import Control.Concurrent.MVar (newEmptyMVar, putMVar, takeMVar) -import qualified Data.ByteString as BS -import Data.Either -import Data.Map (fromList) -import Data.Monoid ((<>)) - -import Kafka.Consumer as C -import Kafka.Metadata as M -import Kafka.Producer as P - +import Control.Concurrent.MVar (newEmptyMVar, putMVar, takeMVar) +import Control.Monad (forM, forM_) +import Control.Monad.Loops +import Data.Either +import Data.Map (fromList) +import Data.Monoid ((<>)) +import Kafka.Consumer +import Kafka.Metadata +import Kafka.Producer import Kafka.TestEnv import Test.Hspec -{-# ANN module ("HLint: ignore Redundant do" :: String) #-} +import qualified Data.ByteString as BS + +{- HLINT ignore "Redundant do" -} spec :: Spec spec = do