| Safe Haskell | None |
|---|---|
| Language | Haskell2010 |
Network.Kafka
Contents
- 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
| Show KafkaState # | |
| Generic KafkaState # | |
| type Rep KafkaState # | |
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. |
data PartitionAndLeader #
Constructors
| PartitionAndLeader | |
Fields
| |
Instances
data TopicAndMessage #
A topic with a serializable message.
Constructors
| TopicAndMessage | |
Fields
| |
Instances
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.