Question about output flow

[AndrejMitrovic]

Hi,

First of, appreciate the work on this nice library! ❤️

I've been exploring using cqrs for a simple use-case.

The use-case is:

• A customer can upload two separate files to our system - could be S3, etc.
• We need to determine when both of the files are uploaded. So we should track when each file is uploaded and only perform an additional action when both are present.
• When both files are uploaded we should send a message to SNS which integrates with another downstream service.

So I considered the following approach:

• An Aggregate with the customer as the key.
• Two separate commands for each type of file that is received. E.g. UploadFileA, UploadFileB.
• Equivalent two events which will update the aggregate. E.g. FileAUploaded, FileBUploaded.
• The aggregate could just contain the two file names. E.g. file_a: Option<String>; file_b: Option<String>;.

And now comes the tricky part. I need to which part of the cqrs system needs to be responsible for publishing a message to SNS once both files were uploaded. Currently I'm doing it with a Query.

I considered having a 3rd command PublishFilesToSNS with an equivalent event. But I don't know which part of the system would be pushing this command. Aggregate.apply() is not the right place. Only the main() routine has access to cqrs.execute, so I don't see a way for any other part of the system to add new commands.

An alternative is to not have a 3rd command, but rather only have a 3rd event BothFilesUploaded. And then Aggregate.handle() would manually fire this event if both UploadFileA and UploadFileB commands were received for a particular aggregate. But this is tricky, because we're no longer comparing just the state of the aggregate, but the state of the aggregate plus the command in flight UploadFileA, UploadFileB`.

But this latter approach is the one I took. Visually:

• User uploads file of type A.
• UploadFileA command fired.
• Aggregate.handle() => Prepare FileAUploaded event.
• FileAUploaded event emitted.
• Aggregator.apply() => Set self.file_a = Some(event.file_name).

And then:

• User uploads file of type B.
• UploadFileB command fired.
• Aggregate.handle() => Prepare FileBUploaded event.
• Aggregate.handle() => Realize both files are ready (tricky). Return [FileBUploaded, BothFilesUploaded events.
• Aggregator.apply() => Processes FileBUploaded event. Sets self.file_b = Some(event.file_name).
• Aggregator.apply() => Ignores BothFilesUploaded event.
• Query.dispatch() => Processes BothFilesUploaded event. Pushes something to SNS.

This is the approach I have. But this looks very dirty to me as Aggregator.handle() needs to do tricky checks with the state of the aggregate and the command that's being processed, plus construct some elaborate event out of it.

Maybe this is better done with Views?

Would love to hear your input. Thank you!

[dave-garred]

Hi @AndrejMitrovic,

I think you've described the usual pattern correctly here. The "tricky" part arises because we return events from the command handler rather than applying them immediately (i.e., when you would add them to a mutable result vector) as many other libraries do. This was intentional to keep the library simple and to better fit with standard Rust patterns.

That being said we've also repeatedly considered moving to the apply-immediately pattern for just this reason.

As a side note, it's best to use CQRS in situations where the business logic is quite complex. For your simple problem there may be better ways to do this.