Merge pull request #124 from felixmulder/sync-example

Add example of synchronous production of messages to examples and README

Author: AlexeyRaga

README.md

@@ -165,6 +165,56 @@ mkMessage k v = ProducerRecord
                   }
 ```
 
+### Synchronous sending of messages
+Because of the asynchronous nature of librdkafka. It 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
+`Kafka.Producer.produceMessage'` function.
+
+```haskell
+produceMessage' :: MonadIO m
+                => KafkaProducer
+                -> ProducerRecord
+                -> (DeliveryReport -> IO ())
+                -> m (Either ImmediateError ())
+```
+
+Using this function, you can provide a callback which will be invoked upon the
+produced message's delivery report. With a little help of `MVar`s or similar,
+you can in fact, create a synchronous-like interface.
+
+```haskell
+sendMessageSync :: MonadIO m
+                => KafkaProducer
+                -> ProducerRecord
+                -> m (Either KafkaError Offset)
+sendMessageSync producer record = liftIO $ do
+  -- Create an empty MVar:
+  var <- newEmptyMVar
+
+  -- Produce the message and use the callback to put the delivery report in the
+  -- MVar:
+  res <- produceMessage' producer record (putMVar var)
+
+  case res of
+    Left (ImmediateError err) ->
+      pure (Left err)
+    Right () -> do
+      -- Flush producer queue to make sure you don't get stuck waiting for the
+      -- message to send:
+      flushProducer producer
+
+      -- Wait for the message's delivery report and map accordingly:
+      takeMVar var >>= return . \case
+        DeliverySuccess _ offset -> Right offset
+        DeliveryFailure _ err    -> Left err
+        NoMessageError err       -> Left err
+```
+
+_Note:_ this is a semi-naive solution as this waits forever (or until
+librdkafka times out). You should make sure that your configuration reflects
+the behavior you want out of this functionality.
+
 # Installation
 
 ## Installing librdkafka

example/ProducerExample.hs

@@ -1,15 +1,19 @@
 {-# LANGUAGE OverloadedStrings #-}
+{-# LANGUAGE LambdaCase #-}
 
 module ProducerExample
 where
 
-import Control.Exception     (bracket)
-import Control.Monad         (forM_)
-import Data.ByteString       (ByteString)
-import Data.ByteString.Char8 (pack)
+import Control.Concurrent.MVar (newEmptyMVar, putMVar, takeMVar)
+import Control.Exception       (bracket)
+import Control.Monad           (forM_)
+import Control.Monad.IO.Class  (MonadIO(..))
+import Data.ByteString         (ByteString)
+import Data.ByteString.Char8   (pack)
 import Data.Monoid
+import Kafka.Consumer          (Offset)
 import Kafka.Producer
-import Data.Text             (Text)
+import Data.Text               (Text)
 
 -- Global producer properties
 producerProps :: ProducerProperties
@@ -69,3 +73,32 @@ sendMessages prod = do
 
   putStrLn "Thank you."
   return $ Right ()
+
+-- | An example for sending messages synchronously using the 'produceMessage''
+--   function
+--
+sendMessageSync :: MonadIO m
+                => KafkaProducer
+                -> ProducerRecord
+                -> m (Either KafkaError Offset)
+sendMessageSync producer record = liftIO $ do
+  -- Create an empty MVar:
+  var <- newEmptyMVar
+
+  -- Produce the message and use the callback to put the delivery report in the
+  -- MVar:
+  res <- produceMessage' producer record (putMVar var)
+
+  case res of
+    Left (ImmediateError err) ->
+      pure (Left err)
+    Right () -> do
+      -- Flush producer queue to make sure you don't get stuck waiting for the
+      -- message to send:
+      flushProducer producer
+
+      -- Wait for the message's delivery report and map accordingly:
+      takeMVar var >>= return . \case
+        DeliverySuccess _ offset -> Right offset
+        DeliveryFailure _ err    -> Left err
+        NoMessageError err       -> Left err