And yes,
DistributedProcessing should be used to distribute you processes across the cluster.
@all
@notxcain ^
I guess I should have brought that up earlier
also do you put all the schedule biz logic in the mapAsync function of runEventWatch?
@dispalt after a lot of thoughts I consider
Schedule in general, i.e if it's not a part of your domain, an anti-pattern, it is here for the sake of backwards compatibility, because we use it in our projects at work. What I would suggest for a greenfield project is to use a view side of an aggregate to perform planned actions, e.g. periodically query all subscription where nextBillingDate < now, and invoice them, instead of adding an entry to schedule and react when it fires.
Question on decoding messages as I have a similar problem in Kafka.
Kafka allows you to send multiple types of messages to one topic on which you can register a handler for example.
The problem is, how do you know which message is of which type in your code base? With Avro you can get back to a correct GenericRecord with a schema, but then to go to the correct Scala class is tricky.
It seems you guys do something here: https://github.com/notxcain/aecor/blob/master/modules/akka-persistence-runtime/src/main/scala/aecor/runtime/akkapersistence/AkkaPersistenceRuntimeActor.scala#L123.
I expect Akka has a similar issue where you sent certain messages using a schema to a certain Actor. How does it know how to decode it back to the correct Scala class?
Kafka allows you to send multiple types of messages to one topic on which you can register a handler for example.
The problem is, how do you know which message is of which type in your code base? With Avro you can get back to a correct GenericRecord with a schema, but then to go to the correct Scala class is tricky.
It seems you guys do something here: https://github.com/notxcain/aecor/blob/master/modules/akka-persistence-runtime/src/main/scala/aecor/runtime/akkapersistence/AkkaPersistenceRuntimeActor.scala#L123.
I expect Akka has a similar issue where you sent certain messages using a schema to a certain Actor. How does it know how to decode it back to the correct Scala class?
@guersam it’s not possible to prevent. It’s not Aecor specific. It’s related to delivery guarantees in general. So you have to decide what is preferred from business standpoint. Send 0 or 1 email by committing before sending, or send 1 or possibly infinite emails by committing after sending? :)
@milanvdm you need some kind of a type tag as a hint which will help you to decide what type to use to decode a message to
@notxcain Thanks for the explanation :) If I understood correctly,
1) the former can be implemented with a command handler with side effect like this https://pavkin.ru/aecor-part-2/#gist93471703 or a Saga that orchestrates the consequent commands, and
2) the latter can be implemented with another event handler that subscribes the original journal, filters interesting events and calls external service unless the correlation id is marked as 'done.'
Are both ways possible with aecor, or am I missing something?
@notxcain How would you distribute the different decoders though in a distributed context? Do you just simply keep a mapping between typetags and their respective decoders in a common library?
How do you handle this issue in Aecor? Or does Akka Persistence handle this for you?
How do you handle this issue in Aecor? Or does Akka Persistence handle this for you?
I have
So I send a
When receiving the event in a different service, I can decode every event back to a GenericRecord with all the correct fields.
But as one topic, contains several subtypes, the other service cannot decode it back to a proper Scala class (to properly aggregate your state depending on each event).
So one solution is to include an TypeTag in every Event sent, use that information to get the right Scala class decoder to get back into the Scala types.
This feels very weird though to have some kind of
UserEvent with different subtypes Created, ...So I send a
Created event to kafka. It first gets encoded using Avro into a GenericRecord and the schema gets registered in the registry.When receiving the event in a different service, I can decode every event back to a GenericRecord with all the correct fields.
But as one topic, contains several subtypes, the other service cannot decode it back to a proper Scala class (to properly aggregate your state depending on each event).
So one solution is to include an TypeTag in every Event sent, use that information to get the right Scala class decoder to get back into the Scala types.
This feels very weird though to have some kind of
Map[TypeTag, Decoder] flying around in every consumer.
I assume a similar problem needs to have been solved with Aecor but I cant find any specifics in the repo. Which makes me think it is or hidden in the protobuf or handled by Akka Persistence.
Am I making sense? :)
Example:
val event = consumer.poll[UserEvent](topicName) // Gives back GenericRecord
def fold(e: UserEvent, aggregate: State) = { e match { case Created => .... } }
fold(event, currentState) // Need for the correct decoding of a GenericRecord back to the correct scala type
To get some inspiration
The only place where Aecor handles it is in boopickel-wire-protocol module. And it uses numeric type tags for. Let's say you have
sealed trait FooEvent
case class FooCreated(name: String) extends FooEvent
case class FooDeleted(reason: String) extends FooEventLet's say you have some encoder: FooEvent => Avro and decoder: Avro => Option[FooEvent]. In pseudo code it would be
val encoder = {
case FooCreated(name) => encodeInt(0) then encodeString(name)
case FooDeleted(reason) => encodeInt(1) then encodeString(reason)
}
val decoder = decodeInt.flatMap {
case 0 => decodeString.map(name => FooCreated(name))
case 1 => decoderString.map(reason => FooDeleted(reason))
case _ => Option.empty
}