Producer side: new EventStore(rabbit, mysql, etc)
- publish
send payload to {AggregateType}AggregateIds create/check exchange Product-{aggregateId} create/check all queues registered in redis with expiration of 30 seconds and with name Product-123123-523455-1232123-1231231-{servicename} send message to Product-{aggregateId}
Listener side: new EventStore(rabbit, mysql, redis, etc)
eventsouce.listen(Product, 'projections')
- constructor:
append to eventsourceServices redis key {projectname, Product} if not exists assert ProductAggregateIds and start consuming
- once message arrives at ProductAggregateIds
service will start consuming the Product-{id}-{thisService} if it is not already. Private method consume()
- once message arrives at Product-{id}-{thisService}
will dispatch to all interested listeners once processed will register ack in the ack table with the eventId, service name, projection type and success 1|0
ProductAggregateIds ^ products (LISTEN) ^ ads (LISTEN) ^ api-legacy (LISTEN)
ProductServicesHealthChecks products > send ping ads > send ping
once here the system is "stable", we know which services are listening
send message 123123-523455-1232123-1231231 ^ assert exchange Product-123123-523455-1232123-1231231 with autodelete true ^ create all queues registered in health checks with expiration of 30 seconds and name Product-123123-523455-1232123-1231231-{servicename}
- Queues that have not received anything within 5 will have its consumers deleted, and within 30 seconds the queue should have no consumers
- Once all queues of a exchange have deleted, the exchange should auto delete
Interested services will consume the ProductAggregateIds queue
message received ^ once a message is received, the service will start consuming the Product-{id}-{service} if it is not already ^ once processed will register ack in the ack table with the eventId and service name ^ Users can bind only one time per aggregate type and projection type
ACKS eventId 12312312-31-231-23-123 service products type projections status ok|fail