Allow EventsourcedProcessor to post-process events #373
Conversation
| @@ -113,7 +120,9 @@ trait EventsourcedProcessor extends EventsourcedWriter[Long, Long] with ActorLog | |||
| case payload if processEvent.isDefinedAt(payload) => | |||
| val currentProcessedEvents = processEvent(payload) | |||
| if (lastSequenceNr > processingProgress) | |||
| processedEvents = processedEvents :+ currentProcessedEvents.map(createEvent(_, lastHandledEvent.customDestinationAggregateIds)) | |||
| processedEvents = processedEvents :+ currentProcessedEvents.map { e => | |||
| postProcessDurableEvent(createEvent(e, lastHandledEvent.customDestinationAggregateIds)) | |||
There was a problem hiding this comment.
This allows implementation classes to overwrite emitterId, vectorTimestamp and processId which must be prevented.
There was a problem hiding this comment.
Taking a look at the fields in DurableEvent I'd say there are maybe more than these 3 fields you mentioned that shouldn't be changed:
case class DurableEvent(
payload: Any,
emitterId: String = DurableEvent.UndefinedEmittedId,
emitterAggregateId: Option[String] = None,
customDestinationAggregateIds: Set[String] = Set(),
systemTimestamp: Long = 0L,
vectorTimestamp: VectorTime = VectorTime.Zero,
processId: String = DurableEvent.UndefinedLogId,
localLogId: String = DurableEvent.UndefinedLogId,
localSequenceNr: Long = DurableEvent.UndefinedSequenceNr,
deliveryId: Option[String] = None,
persistOnEventSequenceNr: Option[Long] = None)Maybe I'm missing some use-cases here but I'm not exactly sure which other fields than customDestinationAggregateIds should be allowed to change in the first place:
payloadis "created" inprocessEvent, no need to allow to post-process itemitterIdcould be allowed to change, even though I don't have a use-case in mind nowsystemTimestampshould be set by theEventLogimplementation as I now learned in the context of Allow EventsourcedProcessor to emit events with custom aggregation IDs #370, right? If so, it shouldn't be allowed to be changed here, either.vectorTimestampshould not change as you saidprocessIdshould not change as you saidlocalLogIdprobably not !?localSequenceNrprobably not !?deliveryIdprobably not !?persistOnEventSequenceNrprobably not !?
The assumption of my initial implementation was "garbage in, garbage out", i.e. the implementor of the processor should know what he/she is doing. Therefore, I also didn't add a test because would be no real assumptions that can be verified (other than e.g. the use-case of changing the customDestinationAggregateId which I could obviously add).
WDYT? If nothing should be allowed to change other than customDestinationAggregateIds (and emitterAggregateId?) maybe it would be sensible to add something specific for that feature.
There was a problem hiding this comment.
Good arguments @danbim, thanks. We definitely need a better specification which DurableEvent fields should be allowed to change and which not. More importantly, both the actor-based and the stream-based event processors should enforce that specification consistently. I created a separate ticket for a common event processing model (#376) so that we can cover that later.
In context of this ticket, I'm fine with your initial proposal that is based on the "garbage in, garbage out" assumption. Let's keep it as an (undocumented) expert feature and document the work results of #376 later. It anyway closely matches what can be done with the stream-based processor at the moment.
Regarding the test, adding one that only changes customDestinationAggregateIds should be sufficient.
| @@ -113,7 +120,9 @@ trait EventsourcedProcessor extends EventsourcedWriter[Long, Long] with ActorLog | |||
| case payload if processEvent.isDefinedAt(payload) => | |||
| val currentProcessedEvents = processEvent(payload) | |||
| if (lastSequenceNr > processingProgress) | |||
| processedEvents = processedEvents :+ currentProcessedEvents.map(createEvent(_, lastHandledEvent.customDestinationAggregateIds)) | |||
| processedEvents = processedEvents :+ currentProcessedEvents.map { e => | |||
| postProcessDurableEvent(createEvent(e, lastHandledEvent.customDestinationAggregateIds)) | |||
There was a problem hiding this comment.
Calling postProcessDurableEvent causes the creation of a new function object for each processed event. Function object creation should be done once and only function application should be repeated. Although we don't have a behavior stack here, this is similar to what an EventsourcedView does for onCommand, onEvent and onSnapshot handlers (see https://github.com/RBMHTechnology/eventuate/blob/master/eventuate-core/src/main/scala/com/rbmhtechnology/eventuate/EventsourcedView.scala#L113). Although not touched by your PR, the same problem exists for processEvent and should be changed accordingly. Such changes are a prerequisite for adding support for event processing behavior changes later, if needed.
|
And please write tests for all changes/additions you make 😃 |
- Closes RBMHTechnology#370 in which an EventsourcedProcessor was not able to define custom routing destination aggregate IDs for the events produced by `processEvent` - Introduces method EventsourcedProcessor.postProcessDurableEvent: DurableEvent => DurableEvent which defaults to identity and allows manipulation of the DurableEvent instance created as a consequence of new events generated by `processEvent`
danbim
force-pushed
the
370_EventsourcedProcessor_customDestinationAggregateIds
branch
from
2264feb
to
e5d3fda
Compare
|
Hey Martin! Just added a test. I'm not sure about the "code style" but I think it covers what should be tested. Cheers |
aggregate IDs for the events produced by
processEventwhich defaults to identity and allows manipulation of the DurableEvent instance created as a
consequence of new events generated by
processEvent