| Safe Haskell | None |
|---|---|
| Language | Haskell2010 |
Network.Kafka
Contents
Synopsis
- type KafkaAddress = (Host, Port)
- data KafkaState = KafkaState {
- _stateName :: KafkaString
- _stateRequiredAcks :: RequiredAcks
- _stateRequestTimeout :: Timeout
- _stateWaitSize :: MinBytes
- _stateBufferSize :: MaxBytes
- _stateWaitTime :: MaxWaitTime
- _stateCorrelationId :: CorrelationId
- _stateBrokers :: Map Leader Broker
- _stateConnections :: Map KafkaAddress (Pool Handle)
- _stateTopicMetadata :: Map TopicName TopicMetadata
- _stateAddresses :: 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)
- type KafkaClientId = KafkaString
- data KafkaClientError
- data KafkaTime
- data PartitionAndLeader = PartitionAndLeader {}
- palTopic :: Lens' PartitionAndLeader TopicName
- palPartition :: Lens' PartitionAndLeader Partition
- palLeader :: Lens' PartitionAndLeader Leader
- data TopicAndPartition = TopicAndPartition {}
- data TopicAndMessage = TopicAndMessage {}
- tamTopic :: Lens' TopicAndMessage TopicName
- tamMessage :: Lens' TopicAndMessage Message
- tamPayload :: TopicAndMessage -> ByteString
- defaultCorrelationId :: CorrelationId
- defaultRequiredAcks :: RequiredAcks
- defaultRequestTimeout :: Timeout
- defaultMinBytes :: MinBytes
- defaultMaxBytes :: MaxBytes
- defaultMaxWaitTime :: MaxWaitTime
- mkKafkaState :: KafkaClientId -> KafkaAddress -> KafkaState
- addKafkaAddress :: KafkaAddress -> KafkaState -> KafkaState
- runKafka :: KafkaState -> StateT KafkaState (ExceptT KafkaClientError IO) a -> IO (Either KafkaClientError a)
- tryKafka :: Kafka m => m a -> m a
- makeRequest :: Kafka m => Handle -> ReqResp (m a) -> m a
- metadata :: Kafka m => MetadataRequest -> m MetadataResponse
- metadata' :: Kafka m => Handle -> MetadataRequest -> m MetadataResponse
- getTopicPartitionLeader :: Kafka m => TopicName -> Partition -> m Broker
- expect :: Kafka m => KafkaClientError -> (a -> Maybe b) -> a -> m b
- brokerPartitionInfo :: Kafka m => TopicName -> m (Set PartitionAndLeader)
- findMetadataOrElse :: Kafka m => [TopicName] -> Getting (Maybe a) KafkaState (Maybe a) -> KafkaClientError -> m a
- protocolTime :: KafkaTime -> Time
- updateMetadatas :: Kafka m => [TopicName] -> m ()
- updateMetadata :: Kafka m => TopicName -> m ()
- updateAllMetadata :: Kafka m => m ()
- withBrokerHandle :: Kafka m => Broker -> (Handle -> m a) -> m a
- withAddressHandle :: Kafka m => KafkaAddress -> (Handle -> m a) -> m a
- broker2address :: Broker -> KafkaAddress
- withAnyHandle :: Kafka m => (Handle -> m a) -> m a
- data PartitionOffsetRequestInfo = PartitionOffsetRequestInfo {}
- getLastOffset :: Kafka m => KafkaTime -> Partition -> TopicName -> m Offset
- getLastOffset' :: Kafka m => Handle -> KafkaTime -> Partition -> TopicName -> m Offset
- offsetRequest :: [(TopicAndPartition, PartitionOffsetRequestInfo)] -> OffsetRequest
Documentation
type KafkaAddress = (Host, Port) #
data KafkaState #
Constructors
| KafkaState | |
Fields
| |
Instances
stateBrokers :: Lens' KafkaState (Map Leader Broker) #
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
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 # | |
| Generic KafkaTime # | |
| type Rep KafkaTime # | |
Defined in Network.Kafka type Rep KafkaTime = D1 (MetaData "KafkaTime" "Network.Kafka" "milena-0.5.2.3-EPtlCrT9mL2wAhWEIFpL" False) (C1 (MetaCons "LatestTime" PrefixI False) (U1 :: * -> *) :+: (C1 (MetaCons "EarliestTime" PrefixI False) (U1 :: * -> *) :+: C1 (MetaCons "OtherTime" PrefixI False) (S1 (MetaSel (Nothing :: Maybe Symbol) NoSourceUnpackedness NoSourceStrictness DecidedLazy) (Rec0 Time)))) | |
data PartitionAndLeader #
Constructors
| PartitionAndLeader | |
Fields
| |
Instances
data TopicAndPartition #
Constructors
| TopicAndPartition | |
Fields | |
Instances
data TopicAndMessage #
A topic with a serializable message.
Constructors
| TopicAndMessage | |
Fields
| |
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 :: * -> * # 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.2.3-EPtlCrT9mL2wAhWEIFpL" 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 #
Get the bytes from the Kafka message, ignoring the topic.
Configuration
defaultCorrelationId :: CorrelationId #
Default: 0
defaultRequiredAcks :: RequiredAcks #
Default: 1
defaultRequestTimeout :: Timeout #
Default: 10000
Default: 0
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.
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
| |
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.