Kafka.Producer.Sync

This module provides a synchronous interface on top of the hw-kafka-client

It works by using MVars managed in two different queues. Each request is sent as soon as there are no other effectively equal Kafka records in-flight. This is done in order to make sure that there is no ambiguity as to which MVar to resolve.

Currently, this implements fair sending. For all requests, the oldest pending request should be sent first.

Sync producer

data SyncKafkaProducer

A producer for sending messages to Kafka and waiting for the DeliveryReport

newSyncProducer :: MonadIO m => ProducerProperties -> m (Either KafkaError SyncKafkaProducer)

Create a new SyncKafkaProducer

Note: since this library wraps the regular hw-kafka-client, please be aware that you should not set the delivery report callback. As it is set internally.

produceRecord :: MonadIO m => SyncKafkaProducer -> ProducerRecord -> m (Either KafkaError ())

Re-exports

data KafkaError

Constructors

KafkaError Text

KafkaInvalidReturnValue

KafkaBadSpecification Text

KafkaResponseError RdKafkaRespErrT

KafkaInvalidConfigurationValue Text

KafkaUnknownConfigurationKey Text

KafkaBadConfiguration

Instances

Eq KafkaError

Defined in Kafka.Types

Methods

(==) :: KafkaError -> KafkaError -> Bool

(/=) :: KafkaError -> KafkaError -> Bool

Show KafkaError

Defined in Kafka.Types

Methods

showsPrec :: Int -> KafkaError -> ShowS

show :: KafkaError -> String

showList :: [KafkaError] -> ShowS

Generic KafkaError

Defined in Kafka.Types

Associated Types

type Rep KafkaError :: Type -> Type

Methods

from :: KafkaError -> Rep KafkaError x

to :: Rep KafkaError x -> KafkaError

Exception KafkaError

Defined in Kafka.Types

Methods

toException :: KafkaError -> SomeException

fromException :: SomeException -> Maybe KafkaError

displayException :: KafkaError -> String

type Rep KafkaError

Defined in Kafka.Types

type Rep KafkaError = D1 (MetaData "KafkaError" "Kafka.Types" "hw-kafka-client-3.0.0-f3f0ff4ee651ea9a164f922cc2bc7402c8529ca1a9367543bbc14e3ba60051d1" False) ((C1 (MetaCons "KafkaError" PrefixI False) (S1 (MetaSel (Nothing :: Maybe Symbol) NoSourceUnpackedness NoSourceStrictness DecidedLazy) (Rec0 Text)) :+: (C1 (MetaCons "KafkaInvalidReturnValue" PrefixI False) (U1 :: Type -> Type) :+: C1 (MetaCons "KafkaBadSpecification" PrefixI False) (S1 (MetaSel (Nothing :: Maybe Symbol) NoSourceUnpackedness NoSourceStrictness DecidedLazy) (Rec0 Text)))) :+: ((C1 (MetaCons "KafkaResponseError" PrefixI False) (S1 (MetaSel (Nothing :: Maybe Symbol) NoSourceUnpackedness NoSourceStrictness DecidedLazy) (Rec0 RdKafkaRespErrT)) :+: C1 (MetaCons "KafkaInvalidConfigurationValue" PrefixI False) (S1 (MetaSel (Nothing :: Maybe Symbol) NoSourceUnpackedness NoSourceStrictness DecidedLazy) (Rec0 Text))) :+: (C1 (MetaCons "KafkaUnknownConfigurationKey" PrefixI False) (S1 (MetaSel (Nothing :: Maybe Symbol) NoSourceUnpackedness NoSourceStrictness DecidedLazy) (Rec0 Text)) :+: C1 (MetaCons "KafkaBadConfiguration" PrefixI False) (U1 :: Type -> Type))))

data ProducerRecord

Constructors

ProducerRecord

Fields

• prTopic :: !TopicName
• prPartition :: !ProducePartition
• prKey :: Maybe ByteString
• prValue :: Maybe ByteString

Instances

Eq ProducerRecord

Defined in Kafka.Producer.Types

Methods

(==) :: ProducerRecord -> ProducerRecord -> Bool

(/=) :: ProducerRecord -> ProducerRecord -> Bool

Show ProducerRecord

Defined in Kafka.Producer.Types

Methods

showsPrec :: Int -> ProducerRecord -> ShowS

show :: ProducerRecord -> String

showList :: [ProducerRecord] -> ShowS

Generic ProducerRecord

Defined in Kafka.Producer.Types

Associated Types

type Rep ProducerRecord :: Type -> Type

Methods

from :: ProducerRecord -> Rep ProducerRecord x

to :: Rep ProducerRecord x -> ProducerRecord

type Rep ProducerRecord

Defined in Kafka.Producer.Types

type Rep ProducerRecord = D1 (MetaData "ProducerRecord" "Kafka.Producer.Types" "hw-kafka-client-3.0.0-f3f0ff4ee651ea9a164f922cc2bc7402c8529ca1a9367543bbc14e3ba60051d1" False) (C1 (MetaCons "ProducerRecord" PrefixI True) ((S1 (MetaSel (Just "prTopic") NoSourceUnpackedness SourceStrict DecidedStrict) (Rec0 TopicName) :*: S1 (MetaSel (Just "prPartition") NoSourceUnpackedness SourceStrict DecidedStrict) (Rec0 ProducePartition)) :*: (S1 (MetaSel (Just "prKey") NoSourceUnpackedness NoSourceStrictness DecidedLazy) (Rec0 (Maybe ByteString)) :*: S1 (MetaSel (Just "prValue") NoSourceUnpackedness NoSourceStrictness DecidedLazy) (Rec0 (Maybe ByteString)))))

data ProducePartition

Constructors

SpecifiedPartition !Int

UnassignedPartition

Instances

Eq ProducePartition

Defined in Kafka.Producer.Types

Methods

(==) :: ProducePartition -> ProducePartition -> Bool

(/=) :: ProducePartition -> ProducePartition -> Bool

Ord ProducePartition

Defined in Kafka.Producer.Types

Methods

compare :: ProducePartition -> ProducePartition -> Ordering

(<) :: ProducePartition -> ProducePartition -> Bool

(<=) :: ProducePartition -> ProducePartition -> Bool

(>) :: ProducePartition -> ProducePartition -> Bool

(>=) :: ProducePartition -> ProducePartition -> Bool

max :: ProducePartition -> ProducePartition -> ProducePartition

min :: ProducePartition -> ProducePartition -> ProducePartition

Show ProducePartition

Defined in Kafka.Producer.Types

Methods

showsPrec :: Int -> ProducePartition -> ShowS

show :: ProducePartition -> String

showList :: [ProducePartition] -> ShowS

Generic ProducePartition

Defined in Kafka.Producer.Types

Associated Types

type Rep ProducePartition :: Type -> Type

Methods

from :: ProducePartition -> Rep ProducePartition x

to :: Rep ProducePartition x -> ProducePartition

type Rep ProducePartition

Defined in Kafka.Producer.Types

type Rep ProducePartition = D1 (MetaData "ProducePartition" "Kafka.Producer.Types" "hw-kafka-client-3.0.0-f3f0ff4ee651ea9a164f922cc2bc7402c8529ca1a9367543bbc14e3ba60051d1" False) (C1 (MetaCons "SpecifiedPartition" PrefixI False) (S1 (MetaSel (Nothing :: Maybe Symbol) SourceUnpack SourceStrict DecidedUnpack) (Rec0 Int)) :+: C1 (MetaCons "UnassignedPartition" PrefixI False) (U1 :: Type -> Type))

newtype TopicName

Constructors

TopicName

Fields

• unTopicName :: Text

Instances

Eq TopicName

Defined in Kafka.Types

Methods

(==) :: TopicName -> TopicName -> Bool

(/=) :: TopicName -> TopicName -> Bool

Ord TopicName

Defined in Kafka.Types

Methods

compare :: TopicName -> TopicName -> Ordering

(<) :: TopicName -> TopicName -> Bool

(<=) :: TopicName -> TopicName -> Bool

(>) :: TopicName -> TopicName -> Bool

(>=) :: TopicName -> TopicName -> Bool

max :: TopicName -> TopicName -> TopicName

min :: TopicName -> TopicName -> TopicName

Read TopicName

Defined in Kafka.Types

Methods

readsPrec :: Int -> ReadS TopicName

readList :: ReadS [TopicName]

readPrec :: ReadPrec TopicName

readListPrec :: ReadPrec [TopicName]

Show TopicName

Defined in Kafka.Types

Methods

showsPrec :: Int -> TopicName -> ShowS

show :: TopicName -> String

showList :: [TopicName] -> ShowS

Generic TopicName

Defined in Kafka.Types

Associated Types

type Rep TopicName :: Type -> Type

Methods

from :: TopicName -> Rep TopicName x

to :: Rep TopicName x -> TopicName

type Rep TopicName

Defined in Kafka.Types

type Rep TopicName = D1 (MetaData "TopicName" "Kafka.Types" "hw-kafka-client-3.0.0-f3f0ff4ee651ea9a164f922cc2bc7402c8529ca1a9367543bbc14e3ba60051d1" True) (C1 (MetaCons "TopicName" PrefixI True) (S1 (MetaSel (Just "unTopicName") NoSourceUnpackedness NoSourceStrictness DecidedLazy) (Rec0 Text)))