Merge pull request #168 from phile314/fix_delivery_cb

Fix delivery callback (fallout from #161)

Author: AlexeyRaga

src/Kafka/Producer.hs

@@ -117,7 +117,8 @@ newProducer pps = liftIO $ do
   tc <- topicConf (TopicProps $ (ppTopicProps pps))
 
   -- add default delivery report callback
-  deliveryCallback (const mempty) kc
+  let Callback setDeliveryCallback = deliveryCallback (const mempty)
+  setDeliveryCallback kc
 
   -- set callbacks
   forM_ (ppCallbacks pps) (\(Callback setCb) -> setCb kc)

src/Kafka/Producer/Callbacks.hs

@@ -15,7 +15,7 @@ import           Foreign.StablePtr      (castPtrToStablePtr, deRefStablePtr, fre
 import           Kafka.Callbacks        as X
 import           Kafka.Consumer.Types   (Offset(..))
 import           Kafka.Internal.RdKafka (RdKafkaMessageT(..), RdKafkaRespErrT(..), rdKafkaConfSetDrMsgCb)
-import           Kafka.Internal.Setup   (KafkaConf(..), getRdKafkaConf)
+import           Kafka.Internal.Setup   (KafkaConf(..), getRdKafkaConf, Callback(..))
 import           Kafka.Internal.Shared  (kafkaRespErr, readTopic, readKey, readPayload)
 import           Kafka.Producer.Types   (ProducerRecord(..), DeliveryReport(..), ProducePartition(..))
 import           Kafka.Types            (KafkaError(..), TopicName(..))
@@ -27,8 +27,8 @@ import           Kafka.Types            (KafkaError(..), TopicName(..))
 --   callbacks. For callbacks to individual messsages see
 --   'Kafka.Producer.produceMessage\''./
 --
-deliveryCallback :: (DeliveryReport -> IO ()) -> KafkaConf -> IO ()
-deliveryCallback callback kc = rdKafkaConfSetDrMsgCb (getRdKafkaConf kc) realCb
+deliveryCallback :: (DeliveryReport -> IO ()) -> Callback
+deliveryCallback callback = Callback $ \kc -> rdKafkaConfSetDrMsgCb (getRdKafkaConf kc) realCb
   where
     realCb :: t -> Ptr RdKafkaMessageT -> IO ()
     realCb _ mptr =