kafka-client-sync 0.1.0.1 → 0.1.1.0

raw patch · 4 files changed

+52/−18 lines, 4 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

• CHANGELOG.md +9/−1
• kafka-client-sync.cabal +1/−1
• src/Kafka/Producer/Sync.hs +18/−1
• test/Main.hs +24/−15

@@ -1,4 +1,12 @@-# Revision history for hedgehog-servant+# Revision history for kafka-client-sync++## 0.1.1.0 -- 2019-11-12++* Add `closeSyncProducer` function++## 0.1.0.1 -- 2019-11-10++* Fixup of re-exports and haddock  ## 0.1.0.0 -- 2019-11-10 

@@ -1,7 +1,7 @@ name:   kafka-client-sync version:-  0.1.0.1+  0.1.1.0 synopsis:   Synchronous Kafka Client description:

@@ -12,6 +12,7 @@   ( -- * Sync producer     SyncKafkaProducer   , newSyncProducer+  , closeSyncProducer   , produceRecord      -- * Re-exports@@ -55,7 +56,7 @@ import           Data.Maybe (isJust) import           Data.Sequence (Seq(..), (<|), (|>)) import qualified Kafka.Producer as KP (deliveryCallback, flushProducer, newProducer)-import qualified Kafka.Producer as KP (produceMessage, setCallback)+import qualified Kafka.Producer as KP (closeProducer, produceMessage, setCallback) import           Kafka.Producer.ProducerProperties (ProducerProperties(..)) import qualified Kafka.Producer.ProducerProperties as KP (brokersList, logLevel, compression, topicCompression) import qualified Kafka.Producer.ProducerProperties as KP (sendTimeout, extraProps, suppressDisconnectLogs)@@ -65,6 +66,8 @@ import           Kafka.Types (KafkaLogLevel(..), KafkaError(..), TopicName(..), Timeout(..)) import           Kafka.Types (KafkaDebug(..), BrokerAddress(..), KafkaCompressionCodec(..)) +-- | Synchronously produce a record using the specified producer+-- produceRecord :: MonadIO m => SyncKafkaProducer -> ProducerRecord -> m (Either KafkaError ()) produceRecord syncProducer record =   -- Produce our message synchronously, then send pending:@@ -73,6 +76,10 @@  -- | A producer for sending messages to Kafka and waiting for the 'DeliveryReport' --+--   A single producer may be used for the entire application. The underlying+--   library, @librdkafka@, deals very well with concurrent use - this+--   implementation supports that as well.+-- data SyncKafkaProducer = SyncKafkaProducer   { requests :: MVar Requests   , producer :: KafkaProducer@@ -124,6 +131,16 @@       KP.newProducer $ props <> KP.setCallback (KP.deliveryCallback callbackAction)    producer <&> fmap (SyncKafkaProducer reqs)++-- | Close the 'SyncKafkaProducer'+--+--   After invoking this function, the producer should not be used anymore+--+--   Ideally, you would use 'Control.Exception.bracket' in order to make sure+--   that a producer is not re-used once closed.+--+closeSyncProducer :: MonadIO m => SyncKafkaProducer -> m ()+closeSyncProducer SyncKafkaProducer{..} = KP.closeProducer producer  -- | Sends one pending producer record --

@@ -2,6 +2,7 @@  import           Prelude hiding (sequence) +import           Control.Exception (throw, bracket) import           Control.Monad (when) import           Control.Monad.Parallel (sequence) import           Data.Foldable (for_)@@ -9,7 +10,6 @@ import           Data.Text (pack) import           Data.Text.Encoding (encodeUtf8) import           Kafka.Producer.Sync-import           Kafka.Producer  producerProps :: ProducerProperties producerProps = brokersList [BrokerAddress "localhost:9092"]@@ -24,23 +24,32 @@   , prValue = Just . encodeUtf8 . pack . show $ suffix   } +withProducer :: (SyncKafkaProducer -> IO ()) -> IO ()+withProducer action =+  let+    newProducerOrThrow = newSyncProducer producerProps >>= \case+      Left err ->+        throw . userError $ "Couldn't start producer: " <> show err+      Right producer ->+        pure producer+  in+    bracket newProducerOrThrow closeSyncProducer action+ main :: IO () main =-  newSyncProducer producerProps >>= \case-    Left _ -> error "Couldn't create sync producer"-    Right producer -> do-      putStrLn "Running Kafka sync tests"+  withProducer $ \producer -> do+    putStrLn "Running Kafka sync tests" -      let action = produceRecord producer . message-      res <- sequence $-        fmap action [1..100] ++-        fmap action (replicate 1 100)+    let action = produceRecord producer . message+    res <- sequence $+      fmap action [1..100] +++      fmap action (replicate 1 100) -      for_ res $-        either (putStrLn . (<>) "got error: " . show) pure+    for_ res $+      either (putStrLn . (<>) "got error: " . show) pure -      when-        (any isLeft res)-        (error "Couldn't publish messages in parallel")+    when+      (any isLeft res)+      (throw . userError $ "Couldn't publish messages in parallel") -      putStrLn "Successfully published messages"+    putStrLn "Successfully published messages"