feat: make it possible to filter message by origin #223
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,54 @@ | ||
| /* | ||
| * Copyright (C) 2021-2024 Lightbend Inc. <https://www.lightbend.com> | ||
| */ | ||
|
|
||
| package akka.javasdk; | ||
|
|
||
| import akka.javasdk.annotations.Consume; | ||
| import akka.javasdk.eventsourcedentity.EventSourcedEntity; | ||
| import akka.javasdk.keyvalueentity.KeyValueEntity; | ||
|
|
||
| import java.util.Optional; | ||
|
|
||
| public interface OriginAwareContext extends Context { | ||
|
There was a problem hiding this comment. MessageContext (Consumer) and UpdateContext (Views) now extends |
||
|
|
||
|
|
||
| /** | ||
| * When available, this method returns the region where a message was first created. | ||
| * | ||
| * <ul> | ||
| * <li>It returns a non-empty Optional when consuming events from an {@link EventSourcedEntity} | ||
| * or a change updates from a {@link KeyValueEntity}</li> | ||
| * <li>It returns an empty Optional when consuming messages from a topic (see {@link Consume.FromTopic}) | ||
| * or from another service (see {@link Consume.FromServiceStream})</li> | ||
| * </ul> | ||
| * | ||
| * @return the region where a message was first created. When not applicable, it returns an empty Optional. | ||
| */ | ||
| Optional<String> originRegion(); | ||
|
There was a problem hiding this comment. I mentioned this on the runtime side, but maybe I should do it here. In my opinion, it feels more natural to get the originRegion information from the metadata. We can add a helper method extract it, similar to There was a problem hiding this comment.
There was a problem hiding this comment. I think the implementation with meta will also be much simpler because we are passing this information through all the layers from the runtime to the sdk. There was a problem hiding this comment. The reason why I didn't add it to the metadata is because we used to also have JWT and ACL there and we stepped away from it. JWT and ACL is now only present on the endpoints request, which makes more sense. Also, metadata is available in places where it doesn't make sense to have the There was a problem hiding this comment. About
There was a problem hiding this comment.
Good point, but that implies that we have a leaking abstraction because I can run |
||
|
|
||
| /** | ||
| * Returns {@code true} if this message originated in the same region where it is currently being processed. | ||
| * A message is considered to have originated in a region if it was created in that region. | ||
| * In all other regions, the same message is treated as a replication. | ||
| * | ||
| * <p>For messages coming from Akka entities: | ||
| * | ||
| * <ul> | ||
| * <li>If the message is an event from an {@link EventSourcedEntity} or a change update from a | ||
| * {@link KeyValueEntity}, it returns {@code true} if it was originated in the region where this consumer is | ||
| * running. Otherwise, it returns {@code false}. | ||
| * <li>This method will always return {@code false} when consuming messages from another service | ||
| * (see {@link Consume.FromServiceStream}) or from a topic (see {@link Consume.FromTopic}). | ||
| * </ul> | ||
| * | ||
| * @return {@code true} if the message originated in the current processing region, {@code false} otherwise | ||
| */ | ||
| default boolean hasLocalOrigin() { | ||
| // empty means we are consuming from a broker or service-to-service | ||
| // for non-empty origins, it needs to match selfRegion | ||
| // in dev-mode, both will be "" and therefore always considered as local | ||
| return originRegion().stream().allMatch(orig -> orig.equals(selfRegion())); | ||
| } | ||
|
|
||
| } | ||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Context is ubiquitous. All contexts inherit from it. By adding the
selfRegionhere we make it possible to always check 'where' we are.This includes component creation contexts, command contexts, http request context and consumer messages (both consumers and views).