packages feed

kafka-effectful (empty) → 0.1.0.0

raw patch · 11 files changed

+889/−0 lines, 11 filesdep +basedep +bytestringdep +containers

Dependencies added: base, bytestring, containers, effectful-core, hw-kafka-client, text

Files

+ CHANGELOG.md view
@@ -0,0 +1,22 @@+# Changelog++All notable changes to `kafka-effectful` are documented here.++This package follows the [Haskell Package Versioning Policy](https://pvp.haskell.org/).++## 0.1.0.0 — 2026-04-16++Initial release.++This is an experimental release. Breaking changes are expected in+subsequent 0.x versions. Pin to an exact version in production until the+API stabilizes at 1.0.++- `KafkaProducer` effect with `produceMessage` and `flushProducer`+  operations.+- `KafkaConsumer` effect with polling, offset management, partition+  management, and querying operations.+- Resource-safe interpreters that acquire and release Kafka handles+  via `bracket`.+- Errors surfaced through `Effectful.Error.Static` as+  `Error KafkaError`.
+ LICENSE view
@@ -0,0 +1,21 @@+MIT License++Copyright (c) 2026 Nadeem Bitar++Permission is hereby granted, free of charge, to any person obtaining a copy+of this software and associated documentation files (the "Software"), to deal+in the Software without restriction, including without limitation the rights+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell+copies of the Software, and to permit persons to whom the Software is+furnished to do so, subject to the following conditions:++The above copyright notice and this permission notice shall be included in all+copies or substantial portions of the Software.++THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE+SOFTWARE.
+ README.md view
@@ -0,0 +1,100 @@+# kafka-effectful++Effectful effects and interpreters for [hw-kafka-client](https://hackage.haskell.org/package/hw-kafka-client), a Haskell binding to Apache Kafka via librdkafka.++Provides typed, composable `KafkaProducer` and `KafkaConsumer` effects for the [effectful](https://hackage.haskell.org/package/effectful) ecosystem.++> **Status: experimental.** This package is on its first release. The API may change in breaking ways in subsequent 0.x versions. Pin to an exact version in production until 1.0 is tagged.++## Features++- **KafkaProducer** -- send messages and flush the producer queue+- **KafkaConsumer** -- poll messages (single or batch), manage offsets, assign/pause/resume/seek partitions, and query committed offsets, positions, assignments, and subscriptions+- Resource-safe interpreters that acquire and release Kafka handles via `bracket`+- Errors surfaced through `Effectful.Error.Static` (`Error KafkaError`)++## Usage++### Producer++```haskell+import Kafka.Effectful++example :: (IOE :> es, Error KafkaError :> es) => Eff es ()+example =+  runKafkaProducer producerProps $ do+    produceMessage record+    flushProducer+```++### Consumer++```haskell+import Kafka.Effectful++example :: (IOE :> es, Error KafkaError :> es) => Eff es ()+example =+  runKafkaConsumer consumerProps subscription loop+  where+    loop = do+      mbMsg <- pollMessage (Timeout 1000)+      case mbMsg of+        Nothing  -> loop+        Just msg -> do+          commitOffsetMessage OffsetCommit msg+          loop+```++`pollMessage` returns `Nothing` when the timeout elapses without a message+arriving; non-timeout failures are thrown via the `Error KafkaError` effect.++### Running it++The effect handlers `runKafkaProducer` and `runKafkaConsumer` require `IOE` and+`Error KafkaError` in the effect stack. A complete program wires them with+`runEff` and `runError`:++```haskell+{-# LANGUAGE TypeApplications #-}++import Effectful+import Effectful.Error.Static (runError)+import Kafka.Effectful++main :: IO ()+main = do+  result <- runEff . runError @KafkaError $ runProgram+  case result of+    Left (_, err) -> putStrLn ("Kafka error: " <> show err)+    Right ()      -> pure ()+  where+    runProgram =+      runKafkaProducer producerProps $ do+        produceMessage record+        flushProducer+```++Replace `producerProps` and `record` with your own `ProducerProperties` and+`ProducerRecord` values (see the `Kafka.Effectful.Producer` module for the+available builders).++## Module Structure++| Module | Description |+|--------|-------------|+| `Kafka.Effectful` | Convenience re-export of both effects and common types |+| `Kafka.Effectful.Producer` | Producer effect, interpreter, and types |+| `Kafka.Effectful.Consumer` | Consumer effect, interpreter, and types |+| `Kafka.Effectful.Producer.Effect` | `KafkaProducer` effect definition and operations |+| `Kafka.Effectful.Producer.Interpreter` | `runKafkaProducer` interpreter |+| `Kafka.Effectful.Consumer.Effect` | `KafkaConsumer` effect definition and operations |+| `Kafka.Effectful.Consumer.Interpreter` | `runKafkaConsumer` interpreter |++## Requirements++- GHC >= 9.12+- librdkafka (system dependency)++## License++MIT
+ kafka-effectful.cabal view
@@ -0,0 +1,63 @@+cabal-version:   3.4+name:            kafka-effectful+version:         0.1.0.0+synopsis:        Effectful effects for hw-kafka-client+description:+  Effectful effects and interpreters for hw-kafka-client, a Haskell+  binding to Apache Kafka via librdkafka. Provides typed, composable+  KafkaProducer and KafkaConsumer effects.++license:         MIT+license-file:    LICENSE+author:          Nadeem Bitar+maintainer:      Nadeem Bitar+category:        Network, Messaging+build-type:      Simple+extra-doc-files:+  CHANGELOG.md+  README.md++source-repository head+  type:     git+  location: https://github.com/shinzui/kafka-effectful.git++common warnings+  ghc-options:+    -Wall -Wcompat -Widentities -Wincomplete-uni-patterns+    -Wincomplete-record-updates -Wredundant-constraints+    -fhide-source-paths -Wmissing-export-lists -Wpartial-fields+    -Wmissing-deriving-strategies++library+  import:             warnings+  exposed-modules:+    Kafka.Effectful+    Kafka.Effectful.Consumer+    Kafka.Effectful.Consumer.Effect+    Kafka.Effectful.Consumer.Interpreter+    Kafka.Effectful.Producer+    Kafka.Effectful.Producer.Effect+    Kafka.Effectful.Producer.Interpreter++  build-depends:+    , base             >=4.21 && <5+    , bytestring       >=0.11 && <0.13+    , containers       >=0.6  && <0.8+    , effectful-core   >=2.5  && <2.7+    , hw-kafka-client  >=5.3  && <6+    , text             >=2.0  && <2.2++  hs-source-dirs:     src+  default-language:   GHC2024+  default-extensions:+    DataKinds+    DuplicateRecordFields+    GADTs+    GeneralisedNewtypeDeriving+    ImportQualifiedPost+    LambdaCase+    NoFieldSelectors+    OverloadedRecordDot+    OverloadedStrings+    TypeFamilies+    TypeOperators
+ src/Kafka/Effectful.hs view
@@ -0,0 +1,141 @@+module Kafka.Effectful (+    -- * Producer Effect+    KafkaProducer,+    runKafkaProducer,+    produceMessage,+    flushProducer,++    -- * Consumer Effect+    KafkaConsumer,+    runKafkaConsumer,++    -- ** Polling+    pollMessage,+    pollMessageBatch,++    -- ** Offset Management+    commitOffsetMessage,+    commitAllOffsets,+    commitPartitionsOffsets,+    storeOffsets,+    storeOffsetMessage,++    -- ** Partition Management+    assign,+    pausePartitions,+    resumePartitions,+    seekPartitions,++    -- ** Querying+    committed,+    position,+    assignment,+    subscription,++    -- * Producer Types+    ProducerRecord (..),+    ProducePartition (..),+    DeliveryReport (..),+    ImmediateError (..),+    ProducerProperties (..),++    -- * Consumer Types+    ConsumerRecord (..),+    ConsumerProperties (..),+    CallbackPollMode (..),+    Subscription (..),+    Offset (..),+    OffsetReset (..),+    OffsetCommit (..),+    TopicPartition (..),+    SubscribedPartitions (..),+    ConsumerGroupId (..),+    PartitionOffset (..),+    Timestamp (..),+    RebalanceEvent (..),++    -- * Subscription Builders+    topics,+    offsetReset,+    extraSubscriptionProps,++    -- * Common Types+    KafkaError (..),+    TopicName (..),+    BrokerAddress (..),+    Timeout (..),+    BatchSize (..),+    PartitionId (..),+    Millis (..),+    ClientId (..),+    KafkaLogLevel (..),+    KafkaDebug (..),+    KafkaCompressionCodec (..),+    Headers,+    headersFromList,+    headersToList,+)+where++import Kafka.Effectful.Consumer (+    CallbackPollMode (..),+    ConsumerGroupId (..),+    ConsumerProperties (..),+    ConsumerRecord (..),+    KafkaConsumer,+    Offset (..),+    OffsetCommit (..),+    OffsetReset (..),+    PartitionOffset (..),+    RebalanceEvent (..),+    SubscribedPartitions (..),+    Subscription (..),+    Timestamp (..),+    TopicPartition (..),+    assign,+    assignment,+    commitAllOffsets,+    commitOffsetMessage,+    commitPartitionsOffsets,+    committed,+    extraSubscriptionProps,+    offsetReset,+    pausePartitions,+    pollMessage,+    pollMessageBatch,+    position,+    resumePartitions,+    runKafkaConsumer,+    seekPartitions,+    storeOffsetMessage,+    storeOffsets,+    subscription,+    topics,+ )+import Kafka.Effectful.Producer (+    DeliveryReport (..),+    ImmediateError (..),+    KafkaProducer,+    ProducePartition (..),+    ProducerProperties (..),+    ProducerRecord (..),+    flushProducer,+    produceMessage,+    runKafkaProducer,+ )+import Kafka.Types (+    BatchSize (..),+    BrokerAddress (..),+    ClientId (..),+    Headers,+    KafkaCompressionCodec (..),+    KafkaDebug (..),+    KafkaError (..),+    KafkaLogLevel (..),+    Millis (..),+    PartitionId (..),+    Timeout (..),+    TopicName (..),+    headersFromList,+    headersToList,+ )
+ src/Kafka/Effectful/Consumer.hs view
@@ -0,0 +1,101 @@+module Kafka.Effectful.Consumer (+    -- * Effect+    KafkaConsumer,++    -- * Interpreter+    runKafkaConsumer,++    -- * Polling+    pollMessage,+    pollMessageBatch,++    -- * Offset Management+    commitOffsetMessage,+    commitAllOffsets,+    commitPartitionsOffsets,+    storeOffsets,+    storeOffsetMessage,++    -- * Partition Management+    assign,+    pausePartitions,+    resumePartitions,+    seekPartitions,++    -- * Querying+    committed,+    position,+    assignment,+    subscription,++    -- * Consumer Types+    ConsumerRecord (..),+    Offset (..),+    OffsetReset (..),+    OffsetCommit (..),+    TopicPartition (..),+    SubscribedPartitions (..),+    ConsumerGroupId (..),+    PartitionOffset (..),+    Timestamp (..),+    RebalanceEvent (..),++    -- * Configuration+    ConsumerProperties (..),+    CallbackPollMode (..),+    K.brokersList,+    K.autoCommit,+    K.noAutoCommit,+    K.noAutoOffsetStore,+    K.groupId,+    K.clientId,+    K.setCallback,+    K.logLevel,+    K.compression,+    K.suppressDisconnectLogs,+    K.statisticsInterval,+    K.extraProps,+    K.extraProp,+    K.debugOptions,+    K.queuedMaxMessagesKBytes,+    K.callbackPollMode,++    -- * Subscription+    Subscription (..),+    topics,+    offsetReset,+    extraSubscriptionProps,++    -- * Callbacks+    K.rebalanceCallback,+    K.offsetCommitCallback,+    K.errorCallback,+    K.logCallback,+    K.statsCallback,+    K.Callback,++    -- * Common Types+    KafkaError (..),+    TopicName (..),+    BrokerAddress (..),+    Timeout (..),+    BatchSize (..),+    PartitionId (..),+    Millis (..),+    ClientId (..),+    KafkaLogLevel (..),+    KafkaDebug (..),+    KafkaCompressionCodec (..),+    Headers,+    headersFromList,+    headersToList,+)+where++import Kafka.Consumer.ConsumerProperties (CallbackPollMode (..), ConsumerProperties (..))+import Kafka.Consumer.ConsumerProperties qualified as K+import Kafka.Consumer.Subscription (Subscription (..), extraSubscriptionProps, offsetReset, topics)+import Kafka.Consumer.Types (ConsumerGroupId (..), ConsumerRecord (..), Offset (..), OffsetCommit (..), OffsetReset (..), PartitionOffset (..), RebalanceEvent (..), SubscribedPartitions (..), Timestamp (..), TopicPartition (..))+import Kafka.Effectful.Consumer.Effect (KafkaConsumer, assign, assignment, commitAllOffsets, commitOffsetMessage, commitPartitionsOffsets, committed, pausePartitions, pollMessage, pollMessageBatch, position, resumePartitions, seekPartitions, storeOffsetMessage, storeOffsets, subscription)+import Kafka.Effectful.Consumer.Interpreter (runKafkaConsumer)+import Kafka.Types (BatchSize (..), BrokerAddress (..), ClientId (..), Headers, KafkaCompressionCodec (..), KafkaDebug (..), KafkaError (..), KafkaLogLevel (..), Millis (..), PartitionId (..), Timeout (..), TopicName (..), headersFromList, headersToList)
+ src/Kafka/Effectful/Consumer/Effect.hs view
@@ -0,0 +1,188 @@+module Kafka.Effectful.Consumer.Effect (+    -- * Effect+    KafkaConsumer (..),++    -- * Polling+    pollMessage,+    pollMessageBatch,++    -- * Offset Management+    commitOffsetMessage,+    commitAllOffsets,+    commitPartitionsOffsets,+    storeOffsets,+    storeOffsetMessage,++    -- * Partition Management+    assign,+    pausePartitions,+    resumePartitions,+    seekPartitions,++    -- * Querying+    committed,+    position,+    assignment,+    subscription,+)+where++import Data.ByteString (ByteString)+import Data.Map.Strict (Map)+import Effectful (Dispatch (..), DispatchOf, Eff, Effect, (:>))+import Effectful.Dispatch.Dynamic (send)+import Kafka.Consumer.Types (+    ConsumerRecord,+    OffsetCommit,+    SubscribedPartitions,+    TopicPartition,+ )+import Kafka.Types (+    BatchSize,+    KafkaError,+    PartitionId,+    Timeout,+    TopicName,+ )++-- | Effect for Kafka consumer operations.+data KafkaConsumer :: Effect where+    PollMessage ::+        Timeout ->+        KafkaConsumer m (Maybe (ConsumerRecord (Maybe ByteString) (Maybe ByteString)))+    PollMessageBatch ::+        Timeout ->+        BatchSize ->+        KafkaConsumer m [Either KafkaError (ConsumerRecord (Maybe ByteString) (Maybe ByteString))]+    CommitOffsetMessage ::+        OffsetCommit ->+        ConsumerRecord k v ->+        KafkaConsumer m ()+    CommitAllOffsets ::+        OffsetCommit ->+        KafkaConsumer m ()+    CommitPartitionsOffsets ::+        OffsetCommit ->+        [TopicPartition] ->+        KafkaConsumer m ()+    StoreOffsets ::+        [TopicPartition] ->+        KafkaConsumer m ()+    StoreOffsetMessage ::+        ConsumerRecord k v ->+        KafkaConsumer m ()+    Assign ::+        [TopicPartition] ->+        KafkaConsumer m ()+    PausePartitions ::+        [(TopicName, PartitionId)] ->+        KafkaConsumer m ()+    ResumePartitions ::+        [(TopicName, PartitionId)] ->+        KafkaConsumer m ()+    SeekPartitions ::+        [TopicPartition] ->+        Timeout ->+        KafkaConsumer m ()+    Committed ::+        Timeout ->+        [(TopicName, PartitionId)] ->+        KafkaConsumer m [TopicPartition]+    Position ::+        [(TopicName, PartitionId)] ->+        KafkaConsumer m [TopicPartition]+    Assignment ::+        KafkaConsumer m (Map TopicName [PartitionId])+    Subscription ::+        KafkaConsumer m [(TopicName, SubscribedPartitions)]++type instance DispatchOf KafkaConsumer = 'Dynamic++-- Polling++{- | Poll for a single message.++Returns 'Nothing' when the timeout elapses without a message arriving.+Throws 'KafkaError' via the 'Error' effect for any non-timeout failure+(for example, a broker transport error or an assignment revocation).+-}+pollMessage ::+    (KafkaConsumer :> es) =>+    Timeout ->+    Eff es (Maybe (ConsumerRecord (Maybe ByteString) (Maybe ByteString)))+pollMessage = send . PollMessage++-- | Poll for a batch of messages. Per-message errors are preserved in the 'Either'.+pollMessageBatch ::+    (KafkaConsumer :> es) =>+    Timeout ->+    BatchSize ->+    Eff es [Either KafkaError (ConsumerRecord (Maybe ByteString) (Maybe ByteString))]+pollMessageBatch t b = send $ PollMessageBatch t b++-- Offset Management++-- | Commit the offset of a specific message. Throws 'KafkaError' on failure.+commitOffsetMessage ::+    (KafkaConsumer :> es) => OffsetCommit -> ConsumerRecord k v -> Eff es ()+commitOffsetMessage oc cr = send $ CommitOffsetMessage oc cr++-- | Commit offsets for all currently assigned partitions. Throws 'KafkaError' on failure.+commitAllOffsets :: (KafkaConsumer :> es) => OffsetCommit -> Eff es ()+commitAllOffsets = send . CommitAllOffsets++-- | Commit offsets for specific partitions. Throws 'KafkaError' on failure.+commitPartitionsOffsets ::+    (KafkaConsumer :> es) => OffsetCommit -> [TopicPartition] -> Eff es ()+commitPartitionsOffsets oc tps = send $ CommitPartitionsOffsets oc tps++-- | Store offsets locally without committing to the broker. Throws 'KafkaError' on failure.+storeOffsets :: (KafkaConsumer :> es) => [TopicPartition] -> Eff es ()+storeOffsets = send . StoreOffsets++-- | Store a message's offset locally without committing. Throws 'KafkaError' on failure.+storeOffsetMessage :: (KafkaConsumer :> es) => ConsumerRecord k v -> Eff es ()+storeOffsetMessage = send . StoreOffsetMessage++-- Partition Management++-- | Manually assign partitions to the consumer. Throws 'KafkaError' on failure.+assign :: (KafkaConsumer :> es) => [TopicPartition] -> Eff es ()+assign = send . Assign++-- | Pause consuming from the specified partitions. Throws 'KafkaError' on failure.+pausePartitions :: (KafkaConsumer :> es) => [(TopicName, PartitionId)] -> Eff es ()+pausePartitions = send . PausePartitions++-- | Resume consuming from the specified partitions. Throws 'KafkaError' on failure.+resumePartitions :: (KafkaConsumer :> es) => [(TopicName, PartitionId)] -> Eff es ()+resumePartitions = send . ResumePartitions++-- | Seek to specific offsets for partitions. Throws 'KafkaError' on failure.+seekPartitions :: (KafkaConsumer :> es) => [TopicPartition] -> Timeout -> Eff es ()+seekPartitions tps t = send $ SeekPartitions tps t++-- Querying++-- | Get committed offsets for the specified partitions. Throws 'KafkaError' on failure.+committed ::+    (KafkaConsumer :> es) =>+    Timeout ->+    [(TopicName, PartitionId)] ->+    Eff es [TopicPartition]+committed t ps = send $ Committed t ps++-- | Get the current position (last consumed offset + 1). Throws 'KafkaError' on failure.+position ::+    (KafkaConsumer :> es) =>+    [(TopicName, PartitionId)] ->+    Eff es [TopicPartition]+position = send . Position++-- | Get the current partition assignment.+assignment :: (KafkaConsumer :> es) => Eff es (Map TopicName [PartitionId])+assignment = send Assignment++-- | Get the current topic subscription.+subscription :: (KafkaConsumer :> es) => Eff es [(TopicName, SubscribedPartitions)]+subscription = send Subscription
+ src/Kafka/Effectful/Consumer/Interpreter.hs view
@@ -0,0 +1,105 @@+-- The 'EffectHandler' type synonym in effectful-core expands to a+-- constraint that GHC's redundant-constraint check flags on the+-- handler's signature, even though the constraint is required for+-- 'interpret' to type-check. Suppress the warning at the file level.+{-# OPTIONS_GHC -Wno-redundant-constraints #-}++module Kafka.Effectful.Consumer.Interpreter (+    -- * Interpreter+    runKafkaConsumer,+)+where++import Control.Monad (void)+import Data.Foldable (for_)+import Effectful (Eff, IOE, (:>))+import Effectful qualified+import Effectful.Dispatch.Dynamic (EffectHandler, interpret)+import Effectful.Error.Static (Error, throwError)+import Effectful.Exception (ExitCase (..))+import Effectful.Exception qualified as Exception+import Kafka.Consumer (RdKafkaRespErrT (..))+import Kafka.Consumer qualified as K+import Kafka.Consumer.ConsumerProperties (ConsumerProperties)+import Kafka.Consumer.Subscription (Subscription)+import Kafka.Effectful.Consumer.Effect (KafkaConsumer (..))+import Kafka.Types (KafkaError (..))++{- | Run the 'KafkaConsumer' effect.++Acquires a consumer handle from the given properties and subscription,+and releases it when the effect scope ends. Errors are thrown via the+'Error' effect.+-}+runKafkaConsumer ::+    (IOE :> es, Error KafkaError :> es) =>+    ConsumerProperties ->+    Subscription ->+    Eff (KafkaConsumer : es) a ->+    Eff es a+runKafkaConsumer props sub action =+    fst+        <$> Exception.generalBracket+            acquire+            release+            (\consumer -> interpret (handleConsumer consumer) action)+  where+    acquire = do+        result <- Effectful.liftIO $ K.newConsumer props sub+        case result of+            Left err -> throwError err+            Right consumer -> pure consumer++    release consumer = \case+        ExitCaseSuccess _ -> do+            mbErr <- Effectful.liftIO $ K.closeConsumer consumer+            for_ mbErr throwError+        ExitCaseException _ ->+            Effectful.liftIO . void $ K.closeConsumer consumer+        ExitCaseAbort ->+            Effectful.liftIO . void $ K.closeConsumer consumer++handleConsumer ::+    (IOE :> es, Error KafkaError :> es) =>+    K.KafkaConsumer ->+    EffectHandler KafkaConsumer es+handleConsumer consumer _env = \case+    PollMessage timeout -> do+        result <- Effectful.liftIO $ K.pollMessage consumer timeout+        case result of+            Left (KafkaResponseError RdKafkaRespErrTimedOut) -> pure Nothing+            Left err -> throwError err+            Right msg -> pure (Just msg)+    PollMessageBatch timeout batchSize ->+        Effectful.liftIO $ K.pollMessageBatch consumer timeout batchSize+    CommitOffsetMessage oc cr -> throwOnJust $ K.commitOffsetMessage oc consumer cr+    CommitAllOffsets oc -> throwOnJust $ K.commitAllOffsets oc consumer+    CommitPartitionsOffsets oc tps -> throwOnJust $ K.commitPartitionsOffsets oc consumer tps+    StoreOffsets tps -> throwOnJust $ K.storeOffsets consumer tps+    StoreOffsetMessage cr -> throwOnJust $ K.storeOffsetMessage consumer cr+    Assign tps -> throwOnJust $ K.assign consumer tps+    PausePartitions parts ->+        throwOnKafkaErr (K.pausePartitions consumer parts)+    ResumePartitions parts ->+        throwOnKafkaErr (K.resumePartitions consumer parts)+    SeekPartitions tps timeout -> throwOnJust $ K.seekPartitions consumer tps timeout+    Committed timeout parts -> throwOnLeft $ K.committed consumer timeout parts+    Position parts -> throwOnLeft $ K.position consumer parts+    Assignment -> throwOnLeft $ K.assignment consumer+    Subscription -> throwOnLeft $ K.subscription consumer+  where+    throwOnJust action' = do+        mbErr <- Effectful.liftIO action'+        for_ mbErr throwError++    throwOnLeft action' = do+        result <- Effectful.liftIO action'+        case result of+            Left err -> throwError err+            Right a -> pure a++    throwOnKafkaErr action' = do+        err <- Effectful.liftIO action'+        case err of+            KafkaResponseError RdKafkaRespErrNoError -> pure ()+            _ -> throwError err
+ src/Kafka/Effectful/Producer.hs view
@@ -0,0 +1,63 @@+module Kafka.Effectful.Producer (+    -- * Effect+    KafkaProducer,++    -- * Interpreter+    runKafkaProducer,++    -- * Operations+    produceMessage,+    flushProducer,++    -- * Types+    ProducerRecord (..),+    ProducePartition (..),+    DeliveryReport (..),+    ImmediateError (..),++    -- * Configuration+    ProducerProperties (..),+    K.brokersList,+    K.setCallback,+    K.logLevel,+    K.compression,+    K.topicCompression,+    K.sendTimeout,+    K.statisticsInterval,+    K.extraProps,+    K.extraProp,+    K.suppressDisconnectLogs,+    K.extraTopicProps,+    K.debugOptions,++    -- * Callbacks+    K.deliveryCallback,+    K.errorCallback,+    K.logCallback,+    K.statsCallback,+    K.Callback,++    -- * Common Types+    KafkaError (..),+    TopicName (..),+    BrokerAddress (..),+    Timeout (..),+    KafkaLogLevel (..),+    KafkaDebug (..),+    KafkaCompressionCodec (..),+    Headers,+    headersFromList,+    headersToList,+    Offset (..),+)+where++import Kafka.Effectful.Producer.Effect (KafkaProducer, flushProducer, produceMessage)+import Kafka.Effectful.Producer.Interpreter (runKafkaProducer)+import Kafka.Producer.ProducerProperties (ProducerProperties (..))+import Kafka.Producer.ProducerProperties qualified as K+import Kafka.Producer.Types (DeliveryReport (..), ImmediateError (..), ProducePartition (..), ProducerRecord (..))+import Kafka.Types (BrokerAddress (..), Headers, KafkaCompressionCodec (..), KafkaDebug (..), KafkaError (..), KafkaLogLevel (..), Timeout (..), TopicName (..), headersFromList, headersToList)++-- Offset is in Consumer.Types+import Kafka.Consumer.Types (Offset (..))
+ src/Kafka/Effectful/Producer/Effect.hs view
@@ -0,0 +1,30 @@+module Kafka.Effectful.Producer.Effect (+    -- * Effect+    KafkaProducer (..),++    -- * Operations+    produceMessage,+    flushProducer,+)+where++import Effectful (Dispatch (..), DispatchOf, Eff, Effect, (:>))+import Effectful.Dispatch.Dynamic (send)+import Kafka.Producer.Types (ProducerRecord)++-- | Effect for Kafka producer operations.+data KafkaProducer :: Effect where+    ProduceMessage :: ProducerRecord -> KafkaProducer m ()+    FlushProducer :: KafkaProducer m ()++type instance DispatchOf KafkaProducer = 'Dynamic++{- | Send a single message to Kafka.+Throws 'KafkaError' via the 'Error' effect on failure.+-}+produceMessage :: (KafkaProducer :> es) => ProducerRecord -> Eff es ()+produceMessage = send . ProduceMessage++-- | Flush the producer's outbound queue, blocking until all messages are sent.+flushProducer :: (KafkaProducer :> es) => Eff es ()+flushProducer = send FlushProducer
+ src/Kafka/Effectful/Producer/Interpreter.hs view
@@ -0,0 +1,55 @@+-- The 'EffectHandler' type synonym in effectful-core expands to a+-- constraint that GHC's redundant-constraint check flags on the+-- handler's signature, even though the constraint is required for+-- 'interpret' to type-check. Suppress the warning at the file level.+{-# OPTIONS_GHC -Wno-redundant-constraints #-}++module Kafka.Effectful.Producer.Interpreter (+    -- * Interpreter+    runKafkaProducer,+)+where++import Data.Foldable (for_)+import Effectful (Eff, IOE, (:>))+import Effectful qualified+import Effectful.Dispatch.Dynamic (EffectHandler, interpret)+import Effectful.Error.Static (Error, throwError)+import Effectful.Exception qualified as Exception+import Kafka.Effectful.Producer.Effect (KafkaProducer (..))+import Kafka.Producer qualified as K+import Kafka.Producer.ProducerProperties (ProducerProperties)+import Kafka.Types (KafkaError)++{- | Run the 'KafkaProducer' effect.++Acquires a producer handle from the given properties and releases it+when the effect scope ends. Errors are thrown via the 'Error' effect.+-}+runKafkaProducer ::+    (IOE :> es, Error KafkaError :> es) =>+    ProducerProperties ->+    Eff (KafkaProducer : es) a ->+    Eff es a+runKafkaProducer props action =+    Exception.bracket+        acquire+        (Effectful.liftIO . K.closeProducer)+        (\producer -> interpret (handleProducer producer) action)+  where+    acquire = do+        result <- Effectful.liftIO $ K.newProducer props+        case result of+            Left err -> throwError err+            Right producer -> pure producer++handleProducer ::+    (IOE :> es, Error KafkaError :> es) =>+    K.KafkaProducer ->+    EffectHandler KafkaProducer es+handleProducer producer _env = \case+    ProduceMessage record -> do+        mbErr <- Effectful.liftIO $ K.produceMessage producer record+        for_ mbErr throwError+    FlushProducer ->+        Effectful.liftIO $ K.flushProducer producer