Class Reactors

Namespace: Cratis.Chronicle.Reactors

Assembly: Cratis.Chronicle.dll

Represents an implementation of IReactors.

public class Reactors : IReactors, IReactorPartitionRecovery

Inheritance: object

Reactors

Implements: IReactors

IReactorPartitionRecovery

Inherited Members: object.GetType()

object.MemberwiseClone()

object.ToString()

object.Equals(object)

object.Equals(object, object)

object.ReferenceEquals(object, object)

object.GetHashCode()

Extension Methods: ReactorPartitionRecoveryExtensions.ClearFailedPartitionQuarantineFor<TReactor>(IReactors, Partition, bool)

ReactorPartitionRecoveryExtensions.RetryFailedPartitionFor<TReactor>(IReactors, Partition)

ReactorWaitExtensions.WaitForHandlerById(IReactors, ReactorId, TimeSpan?)

ReactorWaitExtensions.WaitForState<TReactor>(IReactors, ObserverRunningState, TimeSpan?)

ReactorWaitExtensions.WaitForThereToBeFailedPartitions<TReactor>(IReactors, TimeSpan?)

ReactorWaitExtensions.WaitTillActive<TReactor>(IReactors, TimeSpan?)

ReactorWaitExtensions.WaitTillReachesEventSequenceNumber<TReactor>(IReactors, EventSequenceNumber, TimeSpan?)

ReactorWaitExtensions.WaitTillSubscribed<TReactor>(IReactors, TimeSpan?)

RegisteredReactorExtensions.GetHandlerFor<T>(IReactors)

RegisteredReactorExtensions.GetStateFor<T>(IReactors)

RegisteredReactorExtensions.Register<T>(IReactors)

RegisteredReactorExtensions.Replay<T>(IReactors)

MethodCalls.CallGenericMethod<TOut, T>(T, Expression<Func<T, Func<TOut>>>, params Type[])

MethodCalls.CallGenericMethod<TOut, T, T1>(T, Expression<Func<T, Func<T1, TOut>>>, T1, params Type[])

MethodCalls.CallGenericMethod<TOut, T, T1, T2>(T, Expression<Func<T, Func<T1, T2, TOut>>>, T1, T2, params Type[])

MethodCalls.CallGenericMethod<TOut, T, T1, T2, T3>(T, Expression<Func<T, Func<T1, T2, T3, TOut>>>, T1, T2, T3, params Type[])

ConverterExtensions.ConvertTo(object, Type)

ConceptExtensions.GetConceptValue(object)

ConceptExtensions.IsConcept(object)

JsonValueExtensions.ToJsonValue(object?)

ObservableExtensions.InvokeAndWrapWithSubject<TResult>(object, Func<CancellationToken, IObservable<TResult>>)

ObservableExtensions.InvokeAndWrapWithTransformSubject<TResult, TSource>(object, Func<CancellationToken, IObservable<TSource>>, Func<TSource, TResult>)

Constructors

Reactors(IEventStore, IEventTypes, IClientArtifactsProvider, IServiceProvider, IClientArtifactsActivator, IActivateReactorMiddlewares, IEventSerializer, ICausationManager, IIdentityProvider, IActivitySource<Reactors>, IReactorSideEffectHandlers, IReactorContextValuesBuilder, IReactorMethodArgumentsResolver, ILogger<Reactors>, ILoggerFactory)

Initializes a new instance of the Reactors class.

public Reactors(IEventStore eventStore, IEventTypes eventTypes, IClientArtifactsProvider clientArtifactsProvider, IServiceProvider serviceProvider, IClientArtifactsActivator artifactActivator, IActivateReactorMiddlewares middlewaresActivator, IEventSerializer eventSerializer, ICausationManager causationManager, IIdentityProvider identityProvider, IActivitySource<Reactors> activitySource, IReactorSideEffectHandlers sideEffectHandlers, IReactorContextValuesBuilder reactorContextValuesBuilder, IReactorMethodArgumentsResolver argumentsResolver, ILogger<Reactors> logger, ILoggerFactory loggerFactory)

Parameters

eventStore IEventStore: IEventStore the Reactors belong to.
eventTypes IEventTypes: IEventTypes for resolving event types.
clientArtifactsProvider IClientArtifactsProvider: IClientArtifactsProvider for getting client artifacts.
serviceProvider IServiceProvider: IServiceProvider to get instances of types.
artifactActivator IClientArtifactsActivator: IClientArtifactsActivator for creating artifact instances.
middlewaresActivator IActivateReactorMiddlewares: IReactorMiddlewares to call.
eventSerializer IEventSerializer: IEventSerializer for serializing of events.
causationManager ICausationManager: ICausationManager for working with causation.
identityProvider IIdentityProvider: IIdentityProvider for managing identity context.
activitySource IActivitySource<Reactors>: IActivitySource<T> for tracing reactor event handling.
sideEffectHandlers IReactorSideEffectHandlers: IReactorSideEffectHandlers for processing return values as side effects.
reactorContextValuesBuilder IReactorContextValuesBuilder: IReactorContextValuesBuilder for resolving append-metadata for side-effect events.
argumentsResolver IReactorMethodArgumentsResolver: IReactorMethodArgumentsResolver for resolving handler method arguments.
logger ILogger<Reactors>: ILogger for logging.
loggerFactory ILoggerFactory: ILoggerFactory for creating loggers.

Methods

ClearFailedPartitionQuarantineFor(Type, Partition, bool)

Clear the quarantine of a failed partition for a registered reactor and reset its retry budget, keeping its attempt history.

public Task<ReactorPartitionQuarantineClearResult> ClearFailedPartitionQuarantineFor(Type reactorType, Partition partition, bool retryImmediately = true)

Parameters

reactorType Type: The registered reactor type.
partition Partition: The quarantined partition.
retryImmediately bool: Whether to start recovering the partition right away.

Returns

Task<ReactorPartitionQuarantineClearResult>: The bounded outcome of the request.

Define(ReactorId, Action<IReactorBuilder>)

Define a reactor fluently, without registering it.

public IReactorDefinition Define(ReactorId id, Action<IReactorBuilder> define)

Parameters

id ReactorId: The stable reactor identifier.
define Action<IReactorBuilder>: Declares the reactor's handlers, event sequence and replay policy on an IReactorBuilder.

Returns

IReactorDefinition: The IReactorDefinition, which can be inspected and then registered with Register(IReactorDefinition).

Remarks

Event types are resolved against the event types this client knows at the time of the call, which is also what an all-events subscription is expanded to.

Exceptions

NoEventTypesForReactor: Thrown when the definition subscribes to no event types.
TypeIsNotAnEventType: Thrown when a typed handler names a type that is not a known event type.

Discover()

Discover all Reactors from the entry assembly and dependencies.

public Task Discover()

Returns

Task: Awaitable task.

FastForward(ReactorId)

Fast forward a specific reactor by its identifier to the tail of the event log, skipping every event it has not yet handled without invoking it.

public Task<FastForwardResult> FastForward(ReactorId reactorId)

Parameters

reactorId ReactorId: ReactorId to fast forward.

Returns

Task<FastForwardResult>: The FastForwardResult describing what happened.

FastForward<TReactor>()

Fast forward a reactor to the tail of the event log, skipping every event it has not yet handled without invoking it.

public Task<FastForwardResult> FastForward<TReactor>() where TReactor : IReactor

Returns

Task<FastForwardResult>: The FastForwardResult describing what happened.

Type Parameters

TReactor: Type of reactor to fast forward.

GetFailedPartitionsFor(Type)

Get any failed partitions for a specific reactor.

public Task<IEnumerable<FailedPartition>> GetFailedPartitionsFor(Type reactorType)

Parameters

reactorType Type: Type of reducer.

Returns

Task<IEnumerable<FailedPartition>>: Collection of FailedPartition, if any.

GetFailedPartitionsFor<TReactor>()

Get any failed partitions for a specific reactor.

public Task<IEnumerable<FailedPartition>> GetFailedPartitionsFor<TReactor>()

Returns

Task<IEnumerable<FailedPartition>>: Collection of FailedPartition, if any.

Type Parameters

TReactor: Type of reducer.

GetHandlerById(ReactorId)

Gets a specific handler by its ReactorId.

public IReactorHandler GetHandlerById(ReactorId id)

Parameters

id ReactorId: ReactorId to get for.

Returns

IReactorHandler: ReactorHandler instance.

GetHandlerFor(Type)

Gets the registered handler for a reactor CLR type.

public IReactorHandler GetHandlerFor(Type reactorType)

Parameters

reactorType Type: The reactor CLR type.

Returns

IReactorHandler: The registered handler.

Exceptions

TypeBasedArtifactRegistrationNotSupported: The custom registry does not support type-based lookup.

GetHandlerFor<TReactor>()

Gets a specific handler by its TReactor type.

public IReactorHandler GetHandlerFor<TReactor>() where TReactor : IReactor

Returns

IReactorHandler: ReactorHandler instance.

Type Parameters

TReactor: The reactor type.

GetStateFor(Type)

Gets a registered reactor's state.

public Task<ReactorState> GetStateFor(Type reactorType)

Parameters

reactorType Type: The reactor CLR type.

Returns

Task<ReactorState>: The reactor state.

Exceptions

TypeBasedArtifactRegistrationNotSupported: The custom registry does not support type-based state lookup.

GetStateFor<TReactor>()

Get the state of a specific reactor.

public Task<ReactorState> GetStateFor<TReactor>() where TReactor : IReactor

Returns

Task<ReactorState>: ReactorState.

Type Parameters

TReactor: Type of reactor get for.

Register()

public Task Register()

Returns

Task: Awaitable task.

Register(IReactorDefinition)

public Task<IReactorHandler> Register(IReactorDefinition definition)

Parameters

definition IReactorDefinition: The IReactorDefinition to register.

Returns

Task<IReactorHandler>: The registered handler, which can be inspected for state and failed partitions.

Remarks

The reactor is an ordinary Chronicle reactor: it has an observer of its own, subscribed to exactly the definition's EventTypes, and it has state, failed partitions and replay like a discovered reactor has. Each delivered event is deserialized to its CLR type and passed to Handle(object, EventContext). The registration is sent to Chronicle immediately, and is sent again whenever the connection is re-established. Delegate handlers report ReactorType as object.

Exceptions

ReactorAlreadyRegistered: Thrown when the identifier already belongs to a reactor in this client.

Register(ReactorId, Action<IReactorBuilder>)

Define and register a reactor fluently, with typed and catch-all handlers.

public Task<IReactorHandler> Register(ReactorId id, Action<IReactorBuilder> define)

Parameters

id ReactorId: The stable reactor identifier.
define Action<IReactorBuilder>: Declares the reactor's handlers, event sequence and replay policy on an IReactorBuilder.

Returns

Task<IReactorHandler>: The registered handler, which can be inspected for state and failed partitions.

Remarks

This is Define(ReactorId, Action<IReactorBuilder>) followed by Register(IReactorDefinition).

Exceptions

NoEventTypesForReactor: Thrown when the definition subscribes to no event types.
TypeIsNotAnEventType: Thrown when a typed handler names a type that is not a known event type.
ReactorAlreadyRegistered: Thrown when the identifier already belongs to a reactor in this client.

Register(ReactorId, Action<IReactorDefinitionBuilder>, Func<ReactorEvent, CancellationToken, Task>)

Registers a reactor from an explicit event subscription and an awaited callback.

public Task<IReactorHandler> Register(ReactorId id, Action<IReactorDefinitionBuilder> configure, Func<ReactorEvent, CancellationToken, Task> handle)

Parameters

id ReactorId: The stable reactor identifier.
configure Action<IReactorDefinitionBuilder>: Configures the event types, sequence, and replay policy.
handle Func<ReactorEvent, CancellationToken, Task>: Handles each delivered event; successful completion acknowledges the event.

Returns

Task<IReactorHandler>: The registered handler, which can be inspected for state and failed partitions.

Remarks

Delegate handlers report ReactorType as object.

Exceptions

NoEventTypesForReactor: Thrown when no event types are configured.
ReactorAlreadyRegistered: Thrown when the identifier already belongs to a reactor in this client.

Register(Type)

Registers a reactor type using this store's declarations.

public Task<IReactorHandler> Register(Type reactorType)

Parameters

reactorType Type: The reactor CLR type.

Returns

Task<IReactorHandler>: The registered handler.

Exceptions

TypeBasedArtifactRegistrationNotSupported: The custom registry does not support type-based registration.

Register<TReactor>()

Registers a TReactor reactor with Chronicle.

public Task<IReactorHandler> Register<TReactor>() where TReactor : IReactor

Returns

Task<IReactorHandler>: Awaitable task.

Type Parameters

TReactor: The reactor type.

Replay(ReactorId)

Replay a specific reactor by its identifier.

public Task<JobId> Replay(ReactorId reactorId)

Parameters

reactorId ReactorId: ReactorId to replay.

Returns

Task<JobId>: The JobId of the replay job that was started or resumed, or NotSet if the reactor is not replayable or its observer is not in a state it can replay from (disconnected, quarantined); reconnect or release it before replaying.

Replay(Type)

Replays a registered reactor CLR type.

public Task<JobId> Replay(Type reactorType)

Parameters

reactorType Type: The reactor CLR type.

Returns

Task<JobId>: The replay job identifier.

Exceptions

TypeBasedArtifactRegistrationNotSupported: The custom registry does not support type-based replay.

Replay<TReactor>()

Replay a specific reactor.

public Task<JobId> Replay<TReactor>() where TReactor : IReactor

Returns

Task<JobId>: The JobId of the replay job that was started or resumed, or NotSet if the reactor is not replayable or its observer is not in a state it can replay from (disconnected, quarantined); reconnect or release it before replaying.

Type Parameters

TReactor: Type of reactor to replay.

RetryFailedPartitionFor(Type, Partition)

Retry a failed partition for a registered reactor.

public Task<ReactorPartitionRetryOutcome> RetryFailedPartitionFor(Type reactorType, Partition partition)

Parameters

reactorType Type: The registered reactor type.
partition Partition: The failed partition.

Returns

Task<ReactorPartitionRetryOutcome>: The bounded outcome of the retry request.

Unregister(ReactorId)

Unregisters a reactor in this client, disconnecting its observation stream. An unknown identifier is ignored.

public void Unregister(ReactorId id)

Parameters

id ReactorId: The reactor identifier to unregister.