hw-kafka-client 4.0.2 → 4.0.3

raw patch · 5 files changed

+14/−11 lines, 5 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

• hw-kafka-client.cabal +1/−1
• src/Kafka/Callbacks.hs +1/−1
• src/Kafka/Consumer.hs +1/−1
• src/Kafka/Internal/RdKafka.chs +10/−7
• src/Kafka/Producer.hs +1/−1

@@ -1,7 +1,7 @@ cabal-version: 2.2  name:                   hw-kafka-client-version:                4.0.2+version:                4.0.3 synopsis:               Kafka bindings for Haskell description:            Apache Kafka bindings backed by the librdkafka C library.                         .

@@ -8,7 +8,7 @@  import Data.ByteString (ByteString) import Kafka.Internal.RdKafka (rdKafkaConfSetErrorCb, rdKafkaConfSetLogCb, rdKafkaConfSetStatsCb)-import Kafka.Internal.Setup (HasKafkaConf(..), getRdKafkaConf, Callback(..))+import Kafka.Internal.Setup (getRdKafkaConf, Callback(..)) import Kafka.Types (KafkaError(..), KafkaLogLevel(..))  -- | Add a callback for errors.

@@ -69,7 +69,7 @@ import           Control.Arrow              (left, (&&&)) import           Control.Concurrent         (forkIO, modifyMVar, rtsSupportsBoundThreads, withMVar) import           Control.Exception          (bracket)-import           Control.Monad              (forM_, mapM_, void, when)+import           Control.Monad              (forM_, void, when) import           Control.Monad.IO.Class     (MonadIO (liftIO)) import           Control.Monad.Trans.Except (ExceptT (ExceptT), runExceptT) import           Data.Bifunctor             (bimap, first)

@@ -11,6 +11,7 @@ import Data.Int (Int32, Int64) import Data.Word (Word8) import Foreign.Concurrent (newForeignPtr)+import qualified Foreign.Concurrent as Concurrent import Foreign.Marshal.Alloc (alloca, allocaBytes) import Foreign.Marshal.Array (peekArray, allocaArray) import Foreign.Storable (Storable(..))@@ -838,9 +839,6 @@     {enumToCInt `RdKafkaTypeT', `RdKafkaConfTPtr', id `CCharBufPointer', cIntConv `CSize'}     -> `RdKafkaTPtr' #} -foreign import ccall safe "rdkafka.h &rd_kafka_destroy"-    rdKafkaDestroy :: FunPtr (Ptr RdKafkaT -> IO ())- newRdKafkaT :: RdKafkaTypeT -> RdKafkaConfTPtr -> IO (Either Text RdKafkaTPtr) newRdKafkaT kafkaType confPtr =     allocaBytes nErrorBytes $ \charPtr -> do@@ -849,10 +847,15 @@         withForeignPtr ret $ \realPtr -> do             if realPtr == nullPtr then peekCText charPtr >>= return . Left             else do-                -- do not call 'rd_kafka_close_consumer' on destroying all Kafka.-                -- when needed, applications should do it explicitly.-                -- {# call rd_kafka_destroy_flags #} realPtr 0x8-                addForeignPtrFinalizer rdKafkaDestroy ret+                -- Issue #151+                -- rd_kafka_destroy_flags may call back into Haskell if an+                -- error or log callback is set, so we must use a concurrent+                -- finalizer+                Concurrent.addForeignPtrFinalizer ret $ do+                  -- do not call 'rd_kafka_close_consumer' on destroying all Kafka.+                  -- when needed, applications should do it explicitly.+                  -- RD_KAFKA_DESTROY_F_NO_CONSUMER_CLOSE = 0x8+                  {# call rd_kafka_destroy_flags #} realPtr 0x8                 return $ Right ret  {#fun rd_kafka_brokers_add as ^

@@ -85,7 +85,7 @@ import           Kafka.Internal.Setup     (Kafka (..), KafkaConf (..), KafkaProps (..), TopicConf (..), TopicProps (..), kafkaConf, topicConf, Callback(..)) import           Kafka.Internal.Shared    (pollEvents) import           Kafka.Producer.Convert   (copyMsgFlags, handleProduceErr', producePartitionCInt, producePartitionInt)-import           Kafka.Producer.Types     (KafkaProducer (..), ImmediateError(..))+import           Kafka.Producer.Types     (KafkaProducer (..))  import Kafka.Producer.ProducerProperties as X import Kafka.Producer.Types              as X hiding (KafkaProducer)