Network.Kafka

Documentation

type KafkaAddress = (Host, Port)

data KafkaState

Constructors

KafkaState

Fields

• _stateName :: KafkaString

  Name to use as a client ID.

• _stateRequiredAcks :: RequiredAcks

  How many acknowledgements are required for producing.

• _stateRequestTimeout :: Timeout

  Time in milliseconds to wait for messages to be produced by broker.

• _stateWaitSize :: MinBytes

  Minimum size of response bytes to block for.

• _stateBufferSize :: MaxBytes

  Maximum size of response bytes to retrieve.

• _stateWaitTime :: MaxWaitTime

  Maximum time in milliseconds to wait for response.

• _stateCorrelationId :: CorrelationId

  An incrementing counter of requests.

• _stateBrokers :: Map Leader Broker

  Broker cache

• _stateConnections :: Map KafkaAddress (Pool Handle)

  Connection cache

• _stateTopicMetadata :: Map TopicName TopicMetadata

  Topic metadata cache

• _stateAddresses :: NonEmpty KafkaAddress

  Address cache

Instances

Show KafkaState

Defined in Network.Kafka

Methods

showsPrec :: Int -> KafkaState -> ShowS

show :: KafkaState -> String

showList :: [KafkaState] -> ShowS

Generic KafkaState

Defined in Network.Kafka

Associated Types

type Rep KafkaState :: Type -> Type

Methods

from :: KafkaState -> Rep KafkaState x

to :: Rep KafkaState x -> KafkaState

type Rep KafkaState

Defined in Network.Kafka

type Rep KafkaState = D1 (MetaData "KafkaState" "Network.Kafka" "milena-0.5.4.0-27nMlnYTid12v1ZBZP3eqt" False) (C1 (MetaCons "KafkaState" PrefixI True) (((S1 (MetaSel (Just "_stateName") NoSourceUnpackedness NoSourceStrictness DecidedLazy) (Rec0 KafkaString) :*: S1 (MetaSel (Just "_stateRequiredAcks") NoSourceUnpackedness NoSourceStrictness DecidedLazy) (Rec0 RequiredAcks)) :*: (S1 (MetaSel (Just "_stateRequestTimeout") NoSourceUnpackedness NoSourceStrictness DecidedLazy) (Rec0 Timeout) :*: (S1 (MetaSel (Just "_stateWaitSize") NoSourceUnpackedness NoSourceStrictness DecidedLazy) (Rec0 MinBytes) :*: S1 (MetaSel (Just "_stateBufferSize") NoSourceUnpackedness NoSourceStrictness DecidedLazy) (Rec0 MaxBytes)))) :*: ((S1 (MetaSel (Just "_stateWaitTime") NoSourceUnpackedness NoSourceStrictness DecidedLazy) (Rec0 MaxWaitTime) :*: (S1 (MetaSel (Just "_stateCorrelationId") NoSourceUnpackedness NoSourceStrictness DecidedLazy) (Rec0 CorrelationId) :*: S1 (MetaSel (Just "_stateBrokers") NoSourceUnpackedness NoSourceStrictness DecidedLazy) (Rec0 (Map Leader Broker)))) :*: (S1 (MetaSel (Just "_stateConnections") NoSourceUnpackedness NoSourceStrictness DecidedLazy) (Rec0 (Map KafkaAddress (Pool Handle))) :*: (S1 (MetaSel (Just "_stateTopicMetadata") NoSourceUnpackedness NoSourceStrictness DecidedLazy) (Rec0 (Map TopicName TopicMetadata)) :*: S1 (MetaSel (Just "_stateAddresses") NoSourceUnpackedness NoSourceStrictness DecidedLazy) (Rec0 (NonEmpty KafkaAddress)))))))

stateWaitTime :: Lens' KafkaState MaxWaitTime

stateWaitSize :: Lens' KafkaState MinBytes

stateTopicMetadata :: Lens' KafkaState (Map TopicName TopicMetadata)

stateRequiredAcks :: Lens' KafkaState RequiredAcks

stateRequestTimeout :: Lens' KafkaState Timeout

stateName :: Lens' KafkaState KafkaString

stateCorrelationId :: Lens' KafkaState CorrelationId

stateConnections :: Lens' KafkaState (Map KafkaAddress (Pool Handle))

stateBufferSize :: Lens' KafkaState MaxBytes

stateBrokers :: Lens' KafkaState (Map Leader Broker)

stateAddresses :: Lens' KafkaState (NonEmpty KafkaAddress)

type Kafka m = (MonadState KafkaState m, MonadError KafkaClientError m, MonadIO m, MonadBaseControl IO m)

The core Kafka monad.

type KafkaClientId = KafkaString

data KafkaClientError

Errors given from the Kafka monad.

Constructors

KafkaNoOffset

A response did not contain an offset.

KafkaDeserializationError String

A value could not be deserialized correctly.

KafkaInvalidBroker Leader

Could not find a cached broker for the found leader.

KafkaFailedToFetchMetadata

KafkaIOException IOException

Instances

Eq KafkaClientError

Defined in Network.Kafka

Methods

(==) :: KafkaClientError -> KafkaClientError -> Bool

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

Show KafkaClientError

Defined in Network.Kafka

Methods

showsPrec :: Int -> KafkaClientError -> ShowS

show :: KafkaClientError -> String

showList :: [KafkaClientError] -> ShowS

Generic KafkaClientError

Defined in Network.Kafka

Associated Types

type Rep KafkaClientError :: Type -> Type

Methods

from :: KafkaClientError -> Rep KafkaClientError x

to :: Rep KafkaClientError x -> KafkaClientError

Exception KafkaClientError

Defined in Network.Kafka

Methods

toException :: KafkaClientError -> SomeException

fromException :: SomeException -> Maybe KafkaClientError

displayException :: KafkaClientError -> String

type Rep KafkaClientError

Defined in Network.Kafka

type Rep KafkaClientError = D1 (MetaData "KafkaClientError" "Network.Kafka" "milena-0.5.4.0-27nMlnYTid12v1ZBZP3eqt" False) ((C1 (MetaCons "KafkaNoOffset" PrefixI False) (U1 :: Type -> Type) :+: C1 (MetaCons "KafkaDeserializationError" PrefixI False) (S1 (MetaSel (Nothing :: Maybe Symbol) NoSourceUnpackedness NoSourceStrictness DecidedLazy) (Rec0 String))) :+: (C1 (MetaCons "KafkaInvalidBroker" PrefixI False) (S1 (MetaSel (Nothing :: Maybe Symbol) NoSourceUnpackedness NoSourceStrictness DecidedLazy) (Rec0 Leader)) :+: (C1 (MetaCons "KafkaFailedToFetchMetadata" PrefixI False) (U1 :: Type -> Type) :+: C1 (MetaCons "KafkaIOException" PrefixI False) (S1 (MetaSel (Nothing :: Maybe Symbol) NoSourceUnpackedness NoSourceStrictness DecidedLazy) (Rec0 IOException)))))

data KafkaTime

An abstract form of Kafka's time. Used for querying offsets.

Constructors

LatestTime

The latest time on the broker.

EarliestTime

The earliest time on the broker.

OtherTime Time

A specific time.

Instances

Eq KafkaTime

Defined in Network.Kafka

Methods

(==) :: KafkaTime -> KafkaTime -> Bool

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

Generic KafkaTime

Defined in Network.Kafka

Associated Types

type Rep KafkaTime :: Type -> Type

Methods

from :: KafkaTime -> Rep KafkaTime x

to :: Rep KafkaTime x -> KafkaTime

type Rep KafkaTime

Defined in Network.Kafka

type Rep KafkaTime = D1 (MetaData "KafkaTime" "Network.Kafka" "milena-0.5.4.0-27nMlnYTid12v1ZBZP3eqt" False) (C1 (MetaCons "LatestTime" PrefixI False) (U1 :: Type -> Type) :+: (C1 (MetaCons "EarliestTime" PrefixI False) (U1 :: Type -> Type) :+: C1 (MetaCons "OtherTime" PrefixI False) (S1 (MetaSel (Nothing :: Maybe Symbol) NoSourceUnpackedness NoSourceStrictness DecidedLazy) (Rec0 Time))))

data PartitionAndLeader

Constructors

PartitionAndLeader

Fields

• _palTopic :: TopicName
• _palPartition :: Partition
• _palLeader :: Leader

Instances

Eq PartitionAndLeader

Defined in Network.Kafka

Methods

(==) :: PartitionAndLeader -> PartitionAndLeader -> Bool

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

Ord PartitionAndLeader

Defined in Network.Kafka

Methods

compare :: PartitionAndLeader -> PartitionAndLeader -> Ordering

(<) :: PartitionAndLeader -> PartitionAndLeader -> Bool

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

(>) :: PartitionAndLeader -> PartitionAndLeader -> Bool

(>=) :: PartitionAndLeader -> PartitionAndLeader -> Bool

max :: PartitionAndLeader -> PartitionAndLeader -> PartitionAndLeader

min :: PartitionAndLeader -> PartitionAndLeader -> PartitionAndLeader

Show PartitionAndLeader

Defined in Network.Kafka

Methods

showsPrec :: Int -> PartitionAndLeader -> ShowS

show :: PartitionAndLeader -> String

showList :: [PartitionAndLeader] -> ShowS

Generic PartitionAndLeader

Defined in Network.Kafka

Associated Types

type Rep PartitionAndLeader :: Type -> Type

Methods

from :: PartitionAndLeader -> Rep PartitionAndLeader x

to :: Rep PartitionAndLeader x -> PartitionAndLeader

type Rep PartitionAndLeader

Defined in Network.Kafka

type Rep PartitionAndLeader = D1 (MetaData "PartitionAndLeader" "Network.Kafka" "milena-0.5.4.0-27nMlnYTid12v1ZBZP3eqt" False) (C1 (MetaCons "PartitionAndLeader" PrefixI True) (S1 (MetaSel (Just "_palTopic") NoSourceUnpackedness NoSourceStrictness DecidedLazy) (Rec0 TopicName) :*: (S1 (MetaSel (Just "_palPartition") NoSourceUnpackedness NoSourceStrictness DecidedLazy) (Rec0 Partition) :*: S1 (MetaSel (Just "_palLeader") NoSourceUnpackedness NoSourceStrictness DecidedLazy) (Rec0 Leader))))

palTopic :: Lens' PartitionAndLeader TopicName

palPartition :: Lens' PartitionAndLeader Partition

palLeader :: Lens' PartitionAndLeader Leader

data TopicAndPartition

Constructors

TopicAndPartition

Fields

• _tapTopic :: TopicName
• _tapPartition :: Partition

Instances

Eq TopicAndPartition

Defined in Network.Kafka

Methods

(==) :: TopicAndPartition -> TopicAndPartition -> Bool

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

Ord TopicAndPartition

Defined in Network.Kafka

Methods

compare :: TopicAndPartition -> TopicAndPartition -> Ordering

(<) :: TopicAndPartition -> TopicAndPartition -> Bool

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

(>) :: TopicAndPartition -> TopicAndPartition -> Bool

(>=) :: TopicAndPartition -> TopicAndPartition -> Bool

max :: TopicAndPartition -> TopicAndPartition -> TopicAndPartition

min :: TopicAndPartition -> TopicAndPartition -> TopicAndPartition

Show TopicAndPartition

Defined in Network.Kafka

Methods

showsPrec :: Int -> TopicAndPartition -> ShowS

show :: TopicAndPartition -> String

showList :: [TopicAndPartition] -> ShowS

Generic TopicAndPartition

Defined in Network.Kafka

Associated Types

type Rep TopicAndPartition :: Type -> Type

Methods

from :: TopicAndPartition -> Rep TopicAndPartition x

to :: Rep TopicAndPartition x -> TopicAndPartition

type Rep TopicAndPartition

Defined in Network.Kafka

type Rep TopicAndPartition = D1 (MetaData "TopicAndPartition" "Network.Kafka" "milena-0.5.4.0-27nMlnYTid12v1ZBZP3eqt" False) (C1 (MetaCons "TopicAndPartition" PrefixI True) (S1 (MetaSel (Just "_tapTopic") NoSourceUnpackedness NoSourceStrictness DecidedLazy) (Rec0 TopicName) :*: S1 (MetaSel (Just "_tapPartition") NoSourceUnpackedness NoSourceStrictness DecidedLazy) (Rec0 Partition)))

data TopicAndMessage

A topic with a serializable message.

Constructors

TopicAndMessage

Fields

• _tamTopic :: TopicName
• _tamMessage :: Message

Instances

Eq TopicAndMessage

Defined in Network.Kafka

Methods

(==) :: TopicAndMessage -> TopicAndMessage -> Bool

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

Show TopicAndMessage

Defined in Network.Kafka

Methods

showsPrec :: Int -> TopicAndMessage -> ShowS

show :: TopicAndMessage -> String

showList :: [TopicAndMessage] -> ShowS

Generic TopicAndMessage

Defined in Network.Kafka

Associated Types

type Rep TopicAndMessage :: Type -> Type

Methods

from :: TopicAndMessage -> Rep TopicAndMessage x

to :: Rep TopicAndMessage x -> TopicAndMessage

type Rep TopicAndMessage

Defined in Network.Kafka

type Rep TopicAndMessage = D1 (MetaData "TopicAndMessage" "Network.Kafka" "milena-0.5.4.0-27nMlnYTid12v1ZBZP3eqt" False) (C1 (MetaCons "TopicAndMessage" PrefixI True) (S1 (MetaSel (Just "_tamTopic") NoSourceUnpackedness NoSourceStrictness DecidedLazy) (Rec0 TopicName) :*: S1 (MetaSel (Just "_tamMessage") NoSourceUnpackedness NoSourceStrictness DecidedLazy) (Rec0 Message)))

tamTopic :: Lens' TopicAndMessage TopicName

tamMessage :: Lens' TopicAndMessage Message

tamPayload :: TopicAndMessage -> ByteString

Get the bytes from the Kafka message, ignoring the topic.

Configuration

defaultCorrelationId :: CorrelationId

Default: 0

defaultRequiredAcks :: RequiredAcks

Default: 1

defaultRequestTimeout :: Timeout

Default: 10000

defaultMinBytes :: MinBytes

Default: 0

defaultMaxBytes :: MaxBytes

Default: 1024 * 1024

defaultMaxWaitTime :: MaxWaitTime

Default: 0

mkKafkaState :: KafkaClientId -> KafkaAddress -> KafkaState

Create a consumer using default values.

addKafkaAddress :: KafkaAddress -> KafkaState -> KafkaState

runKafka :: KafkaState -> StateT KafkaState (ExceptT KafkaClientError IO) a -> IO (Either KafkaClientError a)

Run the underlying Kafka monad.

tryKafka :: Kafka m => m a -> m a

Catch IOExceptions and wrap them in KafkaIOExceptions.

makeRequest :: Kafka m => Handle -> ReqResp (m a) -> m a

Make a request, incrementing the _stateCorrelationId.

metadata :: Kafka m => MetadataRequest -> m MetadataResponse

Send a metadata request to any broker.

metadata' :: Kafka m => Handle -> MetadataRequest -> m MetadataResponse

Send a metadata request.

createTopic :: Kafka m => CreateTopicsRequest -> m CreateTopicsResponse

createTopic' :: Kafka m => Handle -> CreateTopicsRequest -> m CreateTopicsResponse

createTopicsRequest :: TopicName -> Partition -> ReplicationFactor -> [(Partition, Replicas)] -> [(KafkaString, Metadata)] -> CreateTopicsRequest

deleteTopic :: Kafka m => DeleteTopicsRequest -> m DeleteTopicsResponse

deleteTopic' :: Kafka m => Handle -> DeleteTopicsRequest -> m DeleteTopicsResponse

deleteTopicsRequest :: TopicName -> DeleteTopicsRequest

fetchOffset :: Kafka m => OffsetFetchRequest -> m OffsetFetchResponse

fetchOffset' :: Kafka m => Handle -> OffsetFetchRequest -> m OffsetFetchResponse

fetchOffsetRequest :: ConsumerGroup -> TopicName -> Partition -> OffsetFetchRequest

commitOffset :: Kafka m => OffsetCommitRequest -> m OffsetCommitResponse

commitOffset' :: Kafka m => Handle -> OffsetCommitRequest -> m OffsetCommitResponse

commitOffsetRequest :: ConsumerGroup -> TopicName -> Partition -> Offset -> OffsetCommitRequest

getTopicPartitionLeader :: Kafka m => TopicName -> Partition -> m Broker

expect :: Kafka m => KafkaClientError -> (a -> Maybe b) -> a -> m b

brokerPartitionInfo :: Kafka m => TopicName -> m (Set PartitionAndLeader)

Find a leader and partition for the topic.

findMetadataOrElse :: Kafka m => [TopicName] -> Getting (Maybe a) KafkaState (Maybe a) -> KafkaClientError -> m a

protocolTime :: KafkaTime -> Time

Convert an abstract time to a serializable protocol value.

updateMetadatas :: Kafka m => [TopicName] -> m ()

updateMetadata :: Kafka m => TopicName -> m ()

updateAllMetadata :: Kafka m => m ()

withBrokerHandle :: Kafka m => Broker -> (Handle -> m a) -> m a

Execute a Kafka action with a Handle for the given Broker, updating the connections cache if needed.

When the action throws an IOException, it is caught and returned as a KafkaIOException in the Kafka monad.

Note that when the given action throws an exception, any state changes will be discarded. This includes both IOExceptions and exceptions thrown by throwError from Except.

withAddressHandle :: Kafka m => KafkaAddress -> (Handle -> m a) -> m a

Execute a Kafka action with a Handle for the given KafkaAddress, updating the connections cache if needed.

When the action throws an IOException, it is caught and returned as a KafkaIOException in the Kafka monad.

Note that when the given action throws an exception, any state changes will be discarded. This includes both IOExceptions and exceptions thrown by throwError from Except.

broker2address :: Broker -> KafkaAddress

withAnyHandle :: Kafka m => (Handle -> m a) -> m a

Like withAddressHandle, but round-robins the addresses in the KafkaState.

When the action throws an IOException, it is caught and returned as a KafkaIOException in the Kafka monad.

Note that when the given action throws an exception, any state changes will be discarded. This includes both IOExceptions and exceptions thrown by throwError from Except.

Offsets

data PartitionOffsetRequestInfo

Fields to construct an offset request, per topic and partition.

Constructors

PartitionOffsetRequestInfo

Fields

• _kafkaTime :: KafkaTime

  Time to find an offset for.

• _maxNumOffsets :: MaxNumberOfOffsets

  Number of offsets to retrieve.

getLastOffset :: Kafka m => KafkaTime -> Partition -> TopicName -> m Offset

Get the first found offset.

getLastOffset' :: Kafka m => Handle -> KafkaTime -> Partition -> TopicName -> m Offset

Get the first found offset.

offsetRequest :: [(TopicAndPartition, PartitionOffsetRequestInfo)] -> OffsetRequest

Create an offset request.