Akka Projection gRPC can be used for implementing asynchronous event based service-to-service communication. It provides an implementation of an Akka Projection that uses Akka gRPC as underlying transport between event producer and consumer.
Overview
An Entity stores events in its journal in service A.
Consumer in service B starts an Akka Projection which locally reads its offset for service A’s replication stream.
Service B establishes a replication stream from service A.
Events are read from the journal.
Event is emitted to the replication stream.
Event is handled.
Offset is stored.
Producer continues to read new events from the journal and emit to the stream. As an optimization, events can also be published directly from the entity to the producer.
Dependencies
The Akka dependencies are available from Akka’s library repository. To access them there, you need to configure the URL for this repository.
The Protobuf descriptors are defined when the GrpcReadJournalGrpcReadJournal is created. The descriptors are used when deserializing the received events. The protobufDescriptors is a list of the javaDescriptor for the used protobuf messages. It is defined in the ScalaPB generated Proto companion object. Note that GrpcReadJournal should be created with the GrpcReadJournalGrpcReadJournalapplycreate factory method and not from configuration via GrpcReadJournalProvider when using Protobuf serialization.
The R2dbcProjection has support for storing the offset in a relational database using R2DBC.
The above example is using the ShardedDaemonProcess to distribute the instances of the Projection across the cluster. There are alternative ways of running the ProjectionBehavior as described in Running a Projection
How to implement the EventHandler and choose between different processing semantics is described in the R2dbcProjection documentation.
Start from custom offset
By default, a SourceProvider uses the stored offset when starting the Projection. For some cases, especially with Projections over gRPC, it can be useful to adjust the start offset. The consumer might only be interested in new events, or only most recent events.
The start offset can be adjusted by using the EventSourcedProvider.eventsBySlices method that takes an adjustStartOffset function, which is a function from loaded offset (if any) to the adjusted offset that will be used to by the eventsBySlicesQuery.
This can also be useful in combination with Starting from snapshots, which is enabled on the producer side. In that way the consumer can start without any stored offset and only load snapshots for the most recently used entities.
When creating the GrpcReadJournalGrpcReadJournal a gRPC client is created for the target producer. The same GrpcReadJournal instance and its gRPC client should be shared for the same target producer. The code examples above will share the instance between different Projection instances running in the same ActorSystem. The gRPC clients will automatically be closed when the ActorSystem is terminated.
If there is a need to close the gRPC client before ActorSystem termination the close() method of the GrpcReadJournalGrpcReadJournal can be called. After closing the GrpcReadJournal instance cannot be used again.
Producer
Akka Projections gRPC provides the gRPC service implementation that is used by the consumer side. It is created with the EventProducerEventProducer:
Events can be transformed by application specific code on the producer side. The purpose is to be able to have a different public representation from the internal representation (stored in journal). The transformation functions are registered when creating the EventProducer service. Here is an example of one of those transformation functions accessing the projection envelope to include the shopping cart id in the public message type passed to consumers:
sourceimport scala.concurrent.ExecutionContextimport scala.concurrent.Futureimport scala.concurrent.duration._
import scala.util.Failureimport scala.util.Successimport akka.actor.typed.ActorSystemimport akka.grpc.scaladsl.ServerReflectionimport akka.grpc.scaladsl.ServiceHandlerimport akka.http.scaladsl.Httpimport akka.http.scaladsl.model.HttpRequestimport akka.http.scaladsl.model.HttpResponseobjectShoppingCartServer{def start(interface:String,
port:Int,
system:ActorSystem[_],
grpcService: proto.ShoppingCartService,
eventProducerService:PartialFunction[HttpRequest,Future[HttpResponse]]):Unit={implicit val sys:ActorSystem[_]= system
implicit val ec:ExecutionContext=
system.executionContext
val service:HttpRequest=>Future[HttpResponse]=ServiceHandler.concatOrNotFound(
eventProducerService,
proto.ShoppingCartServiceHandler.partial(grpcService),// ServerReflection enabled to support grpcurl without import-path and proto parametersServerReflection.partial(List(proto.ShoppingCartService)))
val bound =Http().newServerAt(interface, port).bind(service).map(_.addToCoordinatedShutdown(3.seconds))
bound.onComplete {caseSuccess(binding)=>
val address = binding.localAddress
system.log.info("Shopping online at gRPC server {}:{}",
address.getHostString,
address.getPort)caseFailure(ex)=>
system.log.error("Failed to bind gRPC endpoint, terminating system", ex)
system.terminate()}}}
This example includes an application specific ShoppingCartService, which is unrelated to Akka Projections gRPC, but it illustrates how to combine the EventProducer service with other gRPC services.
Durable State as source of events
Projections over gRPC requires events, and are therefore typically used with EventSourcedBehavior. DurableStateBehavior can also be the source of events and be used with Projections over gRPC. For DurableStateBehavior you need to use change events as described in Changes from Durable State.
Filters
By default, events from all entities of the given entity type and slice range are emitted from the producer to the consumer. The transformation function on the producer side can omit certain events, but the offsets for these events are still transferred to the consumer, to ensure sequence number validations and offset storage.
Filters can be used when a consumer is only interested in a subset of the entities. The filters can be defined on both the producer side and on the consumer side, and they can be changed at runtime.
By default, all events are emitted, and filters selectively choose what events to filter out. For some of the filters it is useful to first define a ConsumerFilter.excludeAllConsumerFilter.excludeAll filter and then selectively include events. For example to only include events from topics matching topic filters.
Note
The purpose of filters is to toggle if all events for the entity are to be emitted or not. If the purpose is to filter out certain events you should instead use the transformation function of the producer.
Tags
Tags are typically used for the filters, so first an example of how to tag events in the entity. Here, the tag is based on total quantity of the shopping cart, i.e. the state of the cart. The tags are included in the EventEnvelopeEventEnvelope.
Topics are typically used in publish-subscribe systems, such as MQTT Topics. Event filters can be defined as topic match expressions, with topic hierarchies and wildcards according to the MQTT specification.
A topic consists of one or more levels separated by a forward slash, for example:
myhome/groundfloor/livingroom/temperature
The topic of an event is defined by a tag with a t: prefix. See above example of how to set tags in the entity. You should typically tag all events for a certain entity instance with the same topic tag. The tag prefix can be configured:
a leading slash in topic names, e.g. /myhome/groundfloor/livingroom
Single level wildcard: +
The single-level wildcard is represented by the plus symbol (+) and allows the replacement of a single topic level. By subscribing to a topic with a single-level wildcard, any topic that contains an arbitrary string in place of the wildcard will be matched.
myhome/groundfloor/+/temperature will match these topics:
myhome/groundfloor/livingroom/temperature
myhome/groundfloor/kitchen/temperature
but it will not match:
myhome/groundfloor/kitchen/brightness
myhome/firstfloor/kitchen/temperature
myhome/groundfloor/kitchen/fridge/temperature
Multi level wildcard: #
The multi-level wildcard covers multiple topic levels. It is represented by the hash symbol (#) and must be placed as the last character in the topic expression, preceded by a forward slash.
By subscribing to a topic with a multi-level wildcard, you will receive all events of a topic that begins with the pattern before the wildcard character, regardless of the length or depth of the topic. If the topic expression is specified as # alone, all events will be received.
myhome/groundfloor/# will match these topics:
myhome/groundfloor/livingroom/temperature
myhome/groundfloor/kitchen/temperature
myhome/groundfloor/kitchen/brightness
but it will not match:
myhome/firstfloor/kitchen/temperature
Producer defined filter
The producer may define a filter function on the EventProducerSource.
In this example the decision is based on tags, but the filter function can use anything in the EventEnvelopeEventEnvelope parameter or the event itself. Here, the entity sets the tag based on the total quantity of the shopping cart, which requires the full state of the shopping cart and is not known from an individual event.
Topic filters can be defined in similar way, using withTopicProducerFilter.
Note that the purpose of withProducerFilter and withTopicProducerFilter is to toggle if all events for the entity are to be emitted or not. If the purpose is to filter out certain events you should instead use the Transformation.
The producer filter is evaluated before the transformation function, i.e. the event is the original event and not the transformed event.
A producer filter that excludes an event wins over any consumer defined filter, i.e. if the producer filter function returns false the event will not be emitted.
Consumer defined filter
The consumer may define declarative filters that are sent to the producer and evaluated on the producer side before emitting the events.
Consumer filters consists of exclude and include criteria. In short, the exclude criteria are evaluated first and may be overridden by an include criteria. More precisely, they are evaluated according to the following rules:
Exclude criteria are evaluated first.
If no matching exclude criteria the event is emitted.
If an exclude criteria is matching the include criteria are evaluated.
If no matching include criteria the event is discarded.
If matching include criteria the event is emitted.
The exclude criteria can be a combination of:
ExcludeTags - exclude events with any of the given tags
ExcludeRegexEntityIds - exclude events for entities with entity ids matching the given regular expressions
ExcludeEntityIds - exclude events for entities with the given entity ids
To exclude all events you can use ExcludeRegexEntityIds with .*.
The include criteria can be a combination of:
IncludeTopics - include events with any of the given matching topics
IncludeTags - include events with any of the given tags
IncludeRegexEntityIds - include events for entities with entity ids matching the given regular expressions
IncludeEntityIds - include events for entities with the given entity ids
Note that the streamId must match what is used when initializing the GrpcReadJournal, which by default is from the config property akka.projection.grpc.consumer.stream-id.
The filters can be dynamically changed in runtime without restarting the Projections or the GrpcReadJournal. The updates are incremental. For example if you first add an IncludeTags of tag "medium" and then update the filter with another IncludeTags of tag "large", the full filter consists of both "medium" and "large".
The updated filter is kept and remains after restarts of the Projection instances. If the consumer side is running with Akka Cluster the filter is propagated to other nodes in the cluster automatically with Akka Distributed Data. You only have to update at one place and it will be applied to all running Projections with the given streamId.
Warning
The filters will be cleared in case of a full Cluster stop, which means that you need to take care of populating the initial filters at startup.
When the consumer receives an event that is not the first event for the entity, and it hasn’t processed and stored the offset for the preceding event (previous sequence number) a replay of previous events will be triggered. The reason is that the consumer is typically interested in all events for an entity and must process them in the original order. Even though this is completely automatic it can be good to be aware of since it may have a substantial performance impact to replay many events for many entities.
The event replay is triggered “lazily” when a new event with unexpected sequence number is received, but with the ConsumerFilter.IncludeEntityIds it is possible to explicitly define a sequence number from which the replay will start immediately. You have the following choices for the sequence number in the IncludeEntityIds criteria:
if the previously processed sequence number is known, the next (+1) sequence number can be defined
1 can be used to for replaying all events of the entity
0 can be used to not replay events immediately, but they will be replayed lazily as described previously
Any duplicate events are filtered out by the Projection on the consumer side. This deduplication mechanism depends on how long the Projection will keep old offsets. You may have to increase the configuration for this, but that has the drawback of more memory usage.
Authentication and authorization for the producer can be done by implementing a EventProducerInterceptorEventProducerInterceptor and pass it to the grpcServiceHandler method during producer bootstrap. The interceptor is invoked with the stream id and gRPC request metadata for each incoming request and can return a suitable error through GrpcServiceExceptionGrpcServiceException
Each connected consumer will start a eventsBySlices query that will periodically poll and read events from the journal. That means that the journal database will become a bottleneck, unless it can be scaled out, when number of consumers increase.
For the case of many consumers of the same event stream it is recommended to use EventsBySliceFirehoseQueryEventsBySliceFirehoseQuery. The purpose is to share the stream of events from the database and fan out to connected consumer streams. Thereby fewer queries and loading of events from the database.
The producer service itself can easily be scaled out to more instances.
Starting from snapshots
The producer can use snapshots as starting points and thereby reducing number of events that have to be loaded. This can be useful if the consumer start from zero without any previously processed offset or if it has been disconnected for a long while and its offset is far behind.
Then you need to define the snapshot to event transformation function in EventProducerSource.withStartingFromSnapshots when registering the Producer.
Configuration
Consumer configuration
The configuration for the GrpcReadJournal may look like this:
sourceakka.http.server.enable-http2 = on
akka.projection.grpc.consumer {
client {
host ="127.0.0.1"
host = ${?SHOPPING_CART_SERVICE_GRPC_HOST}
port =8101
port = ${?SHOPPING_CART_SERVICE_GRPC_PORT}use-tls =false}
stream-id ="cart"}
The client section in the configuration defines where the producer is running. It is an Akka gRPC configuration with several connection options.
Reference configuration
The following can be overridden in your application.conf for the Projection specific settings:
sourceakka.projection.grpc {
consumer {class="akka.projection.grpc.consumer.GrpcReadJournalProvider"# Note: these settings are only applied when constructing the consumer from config# if creating the GrpcQuerySettings programmatically these settings are ignored# Configuration of gRPC client.# See https://doc.akka.io/libraries/akka-grpc/current/client/configuration.html#by-configuration
client = ${akka.grpc.client."*"}
client {}# Mandatory field identifying the stream to consume/type of entity, must be a stream id# exposed by the producing/publishing side
stream-id =""# Pass these additional request headers as string values in each request to the producer# can be used for example for authorization in combination with an interceptor in the producer.# Example "x-auth-header": "secret"
additional-request-headers {}
filter {
ddata-read-timeout =3s
ddata-write-timeout =3s}
push-destination {# Parallelism per connected push producer
parallelism =100# Disconnect a producer flow if an event is not written to database within this timeout# producer will reconnect, possibly after backoff.
journal-write-timeout =5s}}
producer {# Query plugin for eventsBySlices, such as "akka.persistence.r2dbc.query".
query-plugin-id =""# When using async transformations it can be good to increase this.
transformation-parallelism =1
filter {
replay-parallelism =3# Topic of an event is defined by an event tag with this prefix
topic-tag-prefix ="t:"}# Interval to send keep alives to keep firewalls etc from closing connections with# large pauses between events going through. Set to 0 to disable keep alive messages.## Note: Currently only used by the event producer push
keep-alive-interval =5s}
replication {# Allow edge replicas to connect and replicate updates
accept-edge-replication = off
# Replicated event sourcing from edge sends each event over sharding, in case that delivery# fails or times out, retry this number of times, with an increasing backoff conntrolled by# the min and max backoff settings
edge-replication-delivery-retries =3
edge-replication-delivery-min-backoff =250ms
edge-replication-delivery-max-backoff =5s}}
akka {
actor {
serializers {
akka-projection-grpc-consumer ="akka.projection.grpc.internal.ConsumerSerializer"}
serialization-identifiers {"akka.projection.grpc.internal.ConsumerSerializer"=1558148901}
serialization-bindings {"akka.projection.grpc.internal.DdataConsumerFilterStore$State"= akka-projection-grpc-consumer
"akka.projection.grpc.internal.DdataConsumerFilterStore$ConsumerFilterKey"= akka-projection-grpc-consumer
}}}
Connecting to more than one producer
If you have several Projections that are connecting to different producer services they can be configured as separate GrpcReadJournalGrpcReadJournal configuration sections.
