Bump librdkafka version to 1.6.1

Author: Levashov, Mykyta

scripts/build-librdkafka

@@ -1,6 +1,6 @@
 #!/bin/bash
 
-RDKAFKA_VER="849c066b559950b02e37a69256f0cb7b04381d0e"
+RDKAFKA_VER="1a722553638bba85dbda5050455f7b9a5ef302de"
 
 PRJ=$PWD
 DST="$PRJ/.librdkafka"

src/Kafka/Producer.hs

@@ -79,6 +79,7 @@ import qualified Data.Text                as Text
 import           Foreign.C.String         (withCString)
 import           Foreign.ForeignPtr       (newForeignPtr_, withForeignPtr)
 import           Foreign.Marshal.Array    (withArrayLen)
+import           Foreign.Marshal.Utils    (withMany)
 import           Foreign.Ptr              (Ptr, nullPtr, plusPtr)
 import           Foreign.Storable         (Storable (..))
 import           Foreign.StablePtr        (newStablePtr, castStablePtrToPtr)
@@ -279,13 +280,12 @@ flushProducer kp = liftIO $ do
 ------------------------------------------------------------------------------------
 
 withHeaders :: Headers -> ([RdKafkaVuT] -> IO a) -> IO a
-withHeaders hds handle = go (headersToList hds) []
+withHeaders hds = withMany allocHeader (headersToList hds)
   where
-    go [] acc = handle acc
-    go ((nm, val) : xs) acc =
-        BS.useAsCString nm $ \cnm ->
+    allocHeader (nm, val) f = 
+      BS.useAsCString nm $ \cnm ->
           withBS (Just val) $ \vp vl ->
-            go xs (Header'RdKafkaVu cnm vp (fromIntegral vl) : acc)
+            f $ Header'RdKafkaVu cnm vp (fromIntegral vl)
 
 withBS :: Maybe BS.ByteString -> (Ptr a -> Int -> IO b) -> IO b
 withBS Nothing f = f nullPtr 0

tests-it/Kafka/IntegrationSpec.hs

@@ -10,6 +10,7 @@ import Control.Monad           (forM, forM_, void)
 import Control.Monad.Loops
 import Data.Either
 import Data.Map                (fromList)
+import qualified Data.Set as Set
 import Data.Monoid             ((<>))
 import Kafka.Consumer
 import Kafka.Metadata
@@ -155,7 +156,7 @@ spec = do
     describe "Kafka.Headers.Spec" $ do
         let testHeaders = headersFromList [("a-header-name", "a-header-value"), ("b-header-name", "b-header-value")]
 
-        specWithKafka "Consumer after records with headers are published" consumerProps $ do
+        specWithKafka "Headers consumer/producer" consumerProps $ do
               it "1. sends 2 messages to test topic enriched with headers" $ \(k, prod) -> do
                   void $ receiveMessages k
 
@@ -166,7 +167,7 @@ spec = do
                   (length <$> res) `shouldBe` Right 2
 
                   forM_ res $ \rcs -> 
-                    forM_ rcs ((`shouldBe` testHeaders) . crHeaders)
+                    forM_ rcs ((`shouldBe` Set.fromList (headersToList testHeaders)) . Set.fromList . headersToList . crHeaders)
 
 ----------------------------------------------------------------------------------------------------------------