milena-0.5.4.0: A Kafka client for Haskell.

Safe HaskellNone
LanguageHaskell2010

Network.Kafka

Contents

Synopsis

Documentation

data KafkaState Source #

Constructors

KafkaState 

Fields

Instances
Show KafkaState Source # 
Instance details

Defined in Network.Kafka

Generic KafkaState Source # 
Instance details

Defined in Network.Kafka

Associated Types

type Rep KafkaState :: Type -> Type #

type Rep KafkaState Source # 
Instance details

Defined in Network.Kafka

data KafkaClientError Source #

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 Source # 
Instance details

Defined in Network.Kafka

Show KafkaClientError Source # 
Instance details

Defined in Network.Kafka

Generic KafkaClientError Source # 
Instance details

Defined in Network.Kafka

Associated Types

type Rep KafkaClientError :: Type -> Type #

Exception KafkaClientError Source # 
Instance details

Defined in Network.Kafka

type Rep KafkaClientError Source # 
Instance details

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 Source #

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 Source # 
Instance details

Defined in Network.Kafka

Generic KafkaTime Source # 
Instance details

Defined in Network.Kafka

Associated Types

type Rep KafkaTime :: Type -> Type #

type Rep KafkaTime Source # 
Instance details

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 Source #

Instances
Eq PartitionAndLeader Source # 
Instance details

Defined in Network.Kafka

Ord PartitionAndLeader Source # 
Instance details

Defined in Network.Kafka

Show PartitionAndLeader Source # 
Instance details

Defined in Network.Kafka

Generic PartitionAndLeader Source # 
Instance details

Defined in Network.Kafka

Associated Types

type Rep PartitionAndLeader :: Type -> Type #

type Rep PartitionAndLeader Source # 
Instance details

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))))

data TopicAndPartition Source #

Instances
Eq TopicAndPartition Source # 
Instance details

Defined in Network.Kafka

Ord TopicAndPartition Source # 
Instance details

Defined in Network.Kafka

Show TopicAndPartition Source # 
Instance details

Defined in Network.Kafka

Generic TopicAndPartition Source # 
Instance details

Defined in Network.Kafka

Associated Types

type Rep TopicAndPartition :: Type -> Type #

type Rep TopicAndPartition Source # 
Instance details

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 Source #

A topic with a serializable message.

Instances
Eq TopicAndMessage Source # 
Instance details

Defined in Network.Kafka

Show TopicAndMessage Source # 
Instance details

Defined in Network.Kafka

Generic TopicAndMessage Source # 
Instance details

Defined in Network.Kafka

Associated Types

type Rep TopicAndMessage :: Type -> Type #

type Rep TopicAndMessage Source # 
Instance details

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)))

tamPayload :: TopicAndMessage -> ByteString Source #

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

Configuration

defaultMaxBytes :: MaxBytes Source #

Default: 1024 * 1024

mkKafkaState :: KafkaClientId -> KafkaAddress -> KafkaState Source #

Create a consumer using default values.

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

Catch IOExceptions and wrap them in KafkaIOExceptions.

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

Make a request, incrementing the _stateCorrelationId.

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

Send a metadata request to any broker.

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

Send a metadata request.

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

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

Find a leader and partition for the topic.

protocolTime :: KafkaTime -> Time Source #

Convert an abstract time to a serializable protocol value.

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

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 Source #

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.

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

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 Source #

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

Constructors

PartitionOffsetRequestInfo 

Fields

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

Get the first found offset.

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

Get the first found offset.