Struct nakadion::consumer::Builder

#[non_exhaustive]
pub struct Builder {
    pub subscription_id: Option<SubscriptionId>,
    pub stream_parameters: Option<StreamParameters>,
    pub instrumentation: Option<Instrumentation>,
    pub tick_interval_millis: Option<TickIntervalMillis>,
    pub inactivity_timeout_secs: Option<InactivityTimeoutSecs>,
    pub stream_dead_policy: Option<StreamDeadPolicy>,
    pub warn_stream_stalled_secs: Option<WarnStreamStalledSecs>,
    pub dispatch_strategy: Option<DispatchStrategy>,
    pub commit_strategy: Option<CommitStrategy>,
    pub abort_connect_on_auth_error: Option<AbortConnectOnAuthError>,
    pub abort_connect_on_subscription_not_found: Option<AbortConnectOnSubscriptionNotFound>,
    pub connect_stream_retry_max_delay_secs: Option<ConnectStreamRetryMaxDelaySecs>,
    pub connect_stream_timeout_secs: Option<ConnectStreamTimeoutSecs>,
    pub commit_attempt_timeout_millis: Option<CommitAttemptTimeoutMillis>,
    pub commit_retry_delay_millis: Option<CommitRetryDelayMillis>,
}

Creates a Consumer

This struct configures and creates a consumer. Before a consumer is build the given values will be validated and defaults will be applied.

The Builder can be created and updated from the environment. When updated from the environment only those values will be updated which were not set before.

De-/Serialization

The Builder supports serialization but the instrumentation will never be part of any serialization and therefore default to None

Environment

When initialized/updated from the environment the following environment variable are used which by are by default prefixed with "NAKADION_" or a custom prefix "_":

For Nakadion itself:

• "SUBSCRIPTION_ID"
• "TICK_INTERVAL_MILLIS"
• "INACTIVITY_TIMEOUT_SECS"
• "STREAM_DEAD_POLICY"
• "WARN_STREAM_STALLED_SECS"
• "DISPATCH_STRATEGY"
• "COMMIT_STRATEGY"
• "ABORT_CONNECT_ON_AUTH_ERROR"
• "ABORT_CONNECT_ON_SUBSCRIPTION_NOT_FOUND"
• "CONNECT_STREAM_RETRY_MAX_DELAY_SECS"
• "CONNECT_STREAM_TIMEOUT_SECS"
• "COMMIT_ATTEMPT_TIMEOUT_MILLIS"
• "COMMIT_RETRY_DELAY_MILLIS"

For stream_parameters:

• "MAX_UNCOMMITTED_EVENTS"
• "BATCH_LIMIT"
• "STREAM_LIMIT"
• "BATCH_FLUSH_TIMEOUT_SECS"
• "BATCH_TIMESPAN_SECS"
• "STREAM_TIMEOUT_SECS"
• "COMMIT_TIMEOUT_SECS"

Fields (Non-exhaustive)

Non-exhaustive structs could have additional fields added in future. Therefore, non-exhaustive structs cannot be constructed in external crates using the traditional Struct { .. } syntax; cannot be matched against without a wildcard ..; and struct update syntax will not work.

subscription_id: Option<SubscriptionId>

The SubscriptionId of the subscription to be consumed.

This value must be set.

stream_parameters: Option<StreamParameters>

Parameters that configure the stream to be consumed.

instrumentation: Option<Instrumentation>

The instrumentation to be used to generate metrics

tick_interval_millis: Option<TickIntervalMillis>

The internal tick interval.

This triggers internal notification used to montitor the state of the currently consumed stream.

inactivity_timeout_secs: Option<InactivityTimeoutSecs>

The time after which a stream or partition is considered inactive.

stream_dead_policy: Option<StreamDeadPolicy>

The time after which a stream is considered stuck and has to be aborted.

warn_stream_stalled_secs: Option<WarnStreamStalledSecs>

Emits a warning when no lines were received from Nakadi for the given time.

dispatch_strategy: Option<DispatchStrategy>

Defines how batches are internally dispatched.

This e.g. configures parallelism.

commit_strategy: Option<CommitStrategy>

Defines when to commit cursors.

It is recommended to set this value instead of letting Nakadion determine defaults.

abort_connect_on_auth_error: Option<AbortConnectOnAuthError>

If true abort the consumer when an auth error occurs while connecting to a stream.

abort_connect_on_subscription_not_found: Option<AbortConnectOnSubscriptionNotFound>

If true abort the consumer when a subscription does not exist when connection to a stream.

connect_stream_retry_max_delay_secs: Option<ConnectStreamRetryMaxDelaySecs>

The maximum retry delay between failed attempts to connect to a stream.

connect_stream_timeout_secs: Option<ConnectStreamTimeoutSecs>

The timeout for a request made to Nakadi to connect to a stream.

commit_attempt_timeout_millis: Option<CommitAttemptTimeoutMillis>

The timeout for a request made to Nakadi to commit cursors.

commit_retry_delay_millis: Option<CommitRetryDelayMillis>

The delay between failed attempts to commit cursors.

Methods

impl Builder

pub fn try_from_env() -> Result<Self, Error>

Creates a new Builder from the environment where all the env vars are prefixed with NAKADION_.

pub fn try_from_env_prefixed<T: AsRef<str>>(prefix: T) -> Result<Self, Error>

Creates a new Builder from the environment where all the env vars are prefixed with <prefix>_.

pub fn fill_from_env(&mut self) -> Result<(), Error>

Sets all values that have not been set so far from the environment.

All the env vars are prefixed with NAKADION_.

pub fn fill_from_env_prefixed<T: AsRef<str>>(
    &mut self,
    prefix: T
) -> Result<(), Error>

Sets all values that have not been set so far from the environment.

All the env vars are prefixed with <prefix>_.

pub fn subscription_id(self, subscription_id: SubscriptionId) -> Self

The SubscriptionId of the subscription to be consumed.

This value must be set.

pub fn stream_parameters(self, params: StreamParameters) -> Self

Parameters that configure the stream to be consumed.

pub fn instrumentation(self, instr: Instrumentation) -> Self

The instrumentation to be used to generate metrics

pub fn tick_interval_millis<T: Into<TickIntervalMillis>>(
    self,
    tick_interval: T
) -> Self

The internal tick interval.

This triggers internal notification used to montitor the state of the currently consumed stream.

pub fn inactivity_timeout_secs<T: Into<InactivityTimeoutSecs>>(
    self,
    inactivity_timeout: T
) -> Self

The time after which a stream or partition is considered inactive.

pub fn stream_dead_policy<T: Into<StreamDeadPolicy>>(
    self,
    stream_dead_policy: T
) -> Self

Define when a stream is considered stuck/dead and has to be aborted.

pub fn warn_stream_stalled_secs<T: Into<WarnStreamStalledSecs>>(
    self,
    warn_stream_stalled_secs: T
) -> Self

Emits a warning when no lines were received from Nakadi for the given time.

pub fn dispatch_strategy(self, dispatch_strategy: DispatchStrategy) -> Self

Defines how batches are internally dispatched.

This e.g. configures parallelism.

pub fn commit_strategy(self, commit_strategy: CommitStrategy) -> Self

Defines when to commit cursors.

It is recommended to set this value instead of letting Nakadion determine defaults.

pub fn abort_connect_on_auth_error<T: Into<AbortConnectOnAuthError>>(
    self,
    abort_connect_on_auth_error: T
) -> Self

If true abort the consumer when an auth error occurs while connecting to a stream.

pub fn abort_connect_on_subscription_not_found<T: Into<AbortConnectOnSubscriptionNotFound>>(
    self,
    abort_connect_on_subscription_not_found: T
) -> Self

If true abort the consumer when a subscription does not exist when connection to a stream.

pub fn connect_stream_retry_max_delay_secs<T: Into<ConnectStreamRetryMaxDelaySecs>>(
    self,
    connect_stream_retry_max_delay_secs: T
) -> Self

The maximum retry delay between failed attempts to connect to a stream.

pub fn connect_stream_timeout_secs<T: Into<ConnectStreamTimeoutSecs>>(
    self,
    connect_stream_timeout_secs: T
) -> Self

The timeout for a request made to Nakadi to connect to a stream.

pub fn commit_attempt_timeout_millis<T: Into<CommitAttemptTimeoutMillis>>(
    self,
    commit_attempt_timeout_millis: T
) -> Self

The timeout for a request made to Nakadi to commit cursors.

pub fn commit_retry_delay_millis<T: Into<CommitRetryDelayMillis>>(
    self,
    commit_retry_delay_millis: T
) -> Self

The delay between failed attempts to commit cursors.

pub fn configure_stream_parameters<F>(self, f: F) -> Self where
    F: FnMut(StreamParameters) -> StreamParameters, 

Modify the current StreamParameters with a closure.

If these have not been set StreamParameters::default() will be passed into the closure.

pub fn try_configure_stream_parameters<F>(self, f: F) -> Result<Self, Error> where
    F: FnMut(StreamParameters) -> Result<StreamParameters, Error>, 

Modify the current StreamParameters with a closure.

If these have not been set StreamParameters::default() will be passed into the closure.

If the closure fails, the whole Builder will fail.

pub fn apply_defaults(&mut self)

Applies the defaults to all values that have not been set so far.

Remember that there is no default for a SubscriptionId which must be set otherwise.

pub fn build_with<C, HF, L>(
    &self,
    api_client: C,
    handler_factory: HF,
    logs: L
) -> Result<Consumer, Error> where
    C: NakadionEssentials + Send + Sync + 'static + Clone,
    HF: BatchHandlerFactory,
    HF::Handler: BatchHandler,
    L: LoggingAdapter, 

Create a Consumer

Trait Implementations

impl Clone for Builder

fn clone(&self) -> Builder

Returns a copy of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl Debug for Builder

fn fmt(&self, f: &mut Formatter) -> Result

Formats the value using the given formatter.

impl Default for Builder

fn default() -> Builder

Returns the "default value" for a type.

impl<'de> Deserialize<'de> for Builder

fn deserialize<__D>(__deserializer: __D) -> Result<Self, __D::Error> where
    __D: Deserializer<'de>, 

Deserialize this value from the given Serde deserializer.

impl Serialize for Builder

fn serialize<__S>(&self, __serializer: __S) -> Result<__S::Ok, __S::Error> where
    __S: Serializer, 

Serialize this value into the given Serde serializer.