-
Notifications
You must be signed in to change notification settings - Fork 3
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
* Add support for stream position in event context / metadata. This allows event handlers to see and reason about which offset each event has in the stream. * Dependencies updated * Added the ability to set eventhandler to / from range dynamically.
- Loading branch information
1 parent
1be65dd
commit 7e27362
Showing
25 changed files
with
347 additions
and
49 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
87 changes: 87 additions & 0 deletions
87
Source/Events.Handling/EventHandlerWithRangeSelectorAttribute.cs
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,87 @@ | ||
| // Copyright (c) Dolittle. All rights reserved. | ||
| // Licensed under the MIT license. See LICENSE file in the project root for full license information. | ||
|
|
||
| using System; | ||
| using Dolittle.SDK.Common.Model; | ||
|
|
||
| namespace Dolittle.SDK.Events.Handling; | ||
|
|
||
| /// <summary> | ||
| /// Allows dynamic selection of the range to process. | ||
| /// </summary> | ||
| /// <typeparam name="T"></typeparam> | ||
| public interface IProcessRangeSelector<T> where T : IProcessRangeSelector<T>, new() | ||
| { | ||
| /// <summary> | ||
| /// Get the range to process. | ||
| /// </summary> | ||
| /// <returns></returns> | ||
| public ProcessRange GetRange(); | ||
| } | ||
|
|
||
| /// <summary> | ||
| /// Decorates a class to indicate the Event Handler Id of the Event Handler class. | ||
| /// </summary> | ||
| [AttributeUsage(AttributeTargets.Class)] | ||
| public class EventHandlerWithRangeSelectorAttribute<T> : Attribute, IDecoratedTypeDecorator<EventHandlerModelId> | ||
| where T : IProcessRangeSelector<T>, new() | ||
| { | ||
| readonly EventHandlerId _eventHandlerId; | ||
| readonly string? _alias; | ||
|
|
||
| /// <summary> | ||
| /// Initializes a new instance of the <see cref="EventHandlerAttribute"/> class. | ||
| /// </summary> | ||
| /// <param name="eventHandlerId">The unique identifier of the event handler.</param> | ||
| /// <param name="partitioned">Whether the event handler is partitioned.</param> | ||
| /// <param name="inScope">The scope that the event handler handles events in.</param> | ||
| /// <param name="alias">The alias for the event handler.</param> | ||
| /// <param name="concurrency">How many events can be processed simultaneously</param> | ||
| public EventHandlerWithRangeSelectorAttribute( | ||
| string eventHandlerId, | ||
| bool partitioned = true, | ||
| string? inScope = null, | ||
| string? alias = null, | ||
| int concurrency = 1 | ||
| ) | ||
| { | ||
| _eventHandlerId = eventHandlerId; | ||
| _alias = alias; | ||
| Concurrency = concurrency; | ||
| var selector = new T(); | ||
| var range = selector.GetRange(); | ||
| StartFrom = range.Mode; | ||
| StartAt = range.StartFrom; | ||
| StopAt = range.StopAt; | ||
| if (StartAt.HasValue && StopAt.HasValue && StartAt >= StopAt) | ||
| { | ||
| throw new ArgumentException("StartFromTimestamp must be before StopAtTimestamp"); | ||
| } | ||
|
|
||
| Partitioned = partitioned; | ||
| Scope = inScope ?? ScopeId.Default; | ||
| } | ||
|
|
||
| public int Concurrency { get; set; } | ||
| public ProcessFrom StartFrom { get; } | ||
| public DateTimeOffset? StartAt { get; } | ||
| public DateTimeOffset? StopAt { get; } | ||
|
|
||
| /// <summary> | ||
| /// Gets a value indicating whether this event handler is partitioned. | ||
| /// </summary> | ||
| public bool Partitioned { get; } | ||
|
|
||
| /// <summary> | ||
| /// Gets the <see cref="ScopeId" />. | ||
| /// </summary> | ||
| public ScopeId Scope { get; } | ||
|
|
||
|
|
||
| /// <inheritdoc /> | ||
| public EventHandlerModelId GetIdentifier(Type decoratedType) | ||
| { | ||
| return new(_eventHandlerId, Partitioned, Scope, _alias ?? decoratedType.Name, Concurrency, StartFrom, | ||
| StartAt, StopAt); | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,46 @@ | ||
| // Copyright (c) Dolittle. All rights reserved. | ||
| // Licensed under the MIT license. See LICENSE file in the project root for full license information. | ||
|
|
||
| using System; | ||
|
|
||
| namespace Dolittle.SDK.Events.Handling; | ||
|
|
||
| /// <summary> | ||
| /// Determine what date/time range of events to process. | ||
| /// </summary> | ||
| public record ProcessRange | ||
| { | ||
| /// <summary> | ||
| /// | ||
| /// </summary> | ||
| /// <param name="Mode">Start from the start of the event log or just process new events</param> | ||
| /// <param name="StartFrom">Do not process events from before this. Can be in the future, Letting an event handler take over at a given point. (Optional)</param> | ||
| /// <param name="StopAt">Do not process events that were committed after this. (Optional)</param> | ||
| public ProcessRange(ProcessFrom Mode = ProcessFrom.Earliest, | ||
| DateTimeOffset? StartFrom = null, | ||
| DateTimeOffset? StopAt = null) | ||
| { | ||
| this.Mode = Mode; | ||
| this.StartFrom = StartFrom; | ||
| this.StopAt = StopAt; | ||
| if (StartFrom.HasValue && StopAt.HasValue && StartFrom.Value > StopAt.Value) | ||
| { | ||
| throw new ArgumentException("StartFrom cannot be after StopAt"); | ||
| } | ||
| } | ||
|
|
||
| public ProcessFrom Mode { get; init; } | ||
|
|
||
| /// <summary>Mode: Start from the start of the event log or just process new events</summary> | ||
| public DateTimeOffset? StartFrom { get; init; } | ||
|
|
||
| /// <summary>Do not process events that were committed after this (Optional)</summary> | ||
| public DateTimeOffset? StopAt { get; init; } | ||
|
|
||
| public void Deconstruct(out ProcessFrom Mode, out DateTimeOffset? StartFrom, out DateTimeOffset? StopAt) | ||
| { | ||
| Mode = this.Mode; | ||
| StartFrom = this.StartFrom; | ||
| StopAt = this.StopAt; | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,18 @@ | ||
| // Copyright (c) Dolittle. All rights reserved. | ||
| // Licensed under the MIT license. See LICENSE file in the project root for full license information. | ||
|
|
||
| using Dolittle.SDK.Concepts; | ||
|
|
||
| namespace Dolittle.SDK.Events; | ||
|
|
||
| /// <summary> | ||
| /// Represents the position of the event in a stream. 0-indexed. | ||
| /// </summary> | ||
| public record StreamPosition(ulong Value) : ConceptAs<ulong>(Value) | ||
| { | ||
| /// <summary> | ||
| /// Implicitly convert a <see cref="uint"/> to an <see cref="StreamPosition"/>. | ||
| /// </summary> | ||
| /// <param name="number">The number.</param> | ||
| public static implicit operator StreamPosition(ulong number) => new(number); | ||
| } |
Oops, something went wrong.