DEV Community: Jeremy Pollock

Postman-powered testing of Akka Serverless gRPC APIs

Jeremy Pollock — Thu, 06 Jan 2022 13:39:57 +0000

Over the holidays, 2021, Postman gifted a fine upgrade to its users: beta support for the gRPC protocol in its API platform. As a Product Manager for Lightbend and helping out on its new gRPC native PaaS for building and running APIs and microservices, I was excited, to say the least. In another, recent blog post, I mentioned my desire to leverage UI test-and-try tools for APIs (my time in the REST API world of Mashery and PubNub was the source of such desire). In that same post though, I noted the lack of several important gRPC features, like server reflection and more robust import capabilities, as blockers; hence, my deep dive, in that post, into the CLI tool, Evans.

But just because one builds APIs and services in Akka Serverless, using gRPC as a first class citizen, doesn't mean that one has to rely solely on gRPC tooling. When running your logic on this serverless platform, you can access your developed, and running, APIs through direct HTTP requests. The product has an "always-on" feature called HTTP transcoding, and its default behaviour is to expose every gRPC method as a POST request to a URI of /<fully qualified service name>/<method name> , with that resource accepting a JSON representation of the full protobuf message. One simply has to look at their API definitions, as captured in a protobuf to get the URIs that are needed. Let's check out what that really means!

NOTE

If you just want to go make some API calls in Postman, you can access the demo collection using the below:

Parsing out URIs from a protobuf

Ready to parse, human? To read through a possibly strangely constructed file and deduce URIs that match the above format of /<fully qualified service name>/<method name>? Well, even if you're not, let's learn!

Below is a protobuf file. If you have no idea what protobuf is, you might want to check out this site: https://developers.google.com/protocol-buffers. For purposes of this post, it can be thought of as our API contract. WSDL? OpenAPI specification? Well, like those, yes. The below file defines what API requests can be made, and what the inputs and outputs are Knowing this, we can look at the API contract and divine an API request.

syntax = "proto3";

// This will be part of the fully qualified service name <1>
package com.example;

import "akkaserverless/annotations.proto";
import "google/api/annotations.proto";
import "google/protobuf/empty.proto";

option java_outer_classname = "CounterApi";

message IncreaseValue {
    string counter_id = 1 [(akkaserverless.field).entity_key = true];
    int32 value = 2;
}

message DecreaseValue {
    string counter_id = 1 [(akkaserverless.field).entity_key = true];
    int32 value = 2;
}

message ResetValue {
    string counter_id = 1 [(akkaserverless.field).entity_key = true];
}

message GetCounter {
    string counter_id = 1 [(akkaserverless.field).entity_key = true];
}

message CurrentCounter {
    int32 value = 1;
}

// This will be part of the fully qualified service name, appended to the package name <2>
service CounterService {
    option (akkaserverless.codegen) = {
        value_entity: {
            name: "com.example.domain.Counter"
            entity_type: "counter"
            state: "com.example.domain.CounterState"

        }
    };

    // each of the below can be an URI API call, appended to the fully qualified service name (separated by a forward slash) <3>
    rpc Increase (IncreaseValue) returns (google.protobuf.Empty);
    rpc Decrease (DecreaseValue) returns (google.protobuf.Empty);
    rpc Reset (ResetValue) returns (google.protobuf.Empty);
    rpc GetCurrentCounter (GetCounter) returns (CurrentCounter);
}

Going from top to bottom, let me call out the critical parts, respective of our URI generation exercise.

We pick up the first part of our fully qualified service name in the package line. In this case, as we start building our URI, /com.example starts us off!
The second important line is the service CounterService one; it contains the structure of the API requests that are supported by the API service. To our fully qualified service name string, we will append the CounterService to form /com.example.CounterService. And as such, we're done with that.
Onward to determining the method name. For each of the rpc lines, contained within the service, there will be an URI created, appending method to our fully qualified service name. What this looks like, would be a set of partial URIs:
```
/com.example.CounterService/Increase
/com.example.CounterService/Decrease
/com.example.CounterService/Reset
/com.example.CounterService/GetCurrentCounter
```

We're ready to go! Get me to Postman!

Well, not so fast. We know that the Always-on HTTP Transcoding feature in Akka Serverless is expecting a POST with a body, in JSON format, representing the API inputs. Where do we find these?

As you look back at the part of the protobuf file that had the method names, as marked by the rpc designation, you will notice that immediately after the method name, there is additional text, wrapped in parentheses. This is the input that the API method in question is expecting. Those input names, e.g. IncreaseValue, map to messages defined further up in the protobuf. And from those messages, we can determine the appropriate JSON to send in our API requests. We simply create a JSON attribute, with values matching the data type specified in the message. For example, message IncreaseValue:

message IncreaseValue {
    string counter_id = 1 [(akkaserverless.field).entity_key = true];
    int32 value = 2;
}

The following JSON could be used as an input to our API request:

{"counter_id": "foo", "value": 1}

Putting the two (URI and body) together, adding the hostname of the running service, we can build a curl command:

curl -XPOST -H "Content-Type: application/json" -d '{"counter_id": "foo", "value": 1}' https://nameless-thunder-9740.us-east1.akkaserverless.app/com.example.CounterService/Increase

Now we're ready for Postman!

Creating HTTP Requests in Postman

Postman has great documentation and is a very intuitive application. So really, I'm not sure I need to lay out the steps to do this. But given that this approach is all about making everyday "normal" HTTP requests to a gRPC service/API, perhaps a quick run-through will be good.

NOTE

When you have to make a choice, blue pill or red...Wait. This isn't The Matrix! But with the recent beta release of gRPC support in Postman, you'll be tempted to pick that path when creating a new request. Don't! Choose HTTP Request!

Let's do this!

If you don't have Postman installed on your machine, head here.
Once you're ready, open up the Postman application.
Click on New.
Click on HTTP Request.
Change HTTP Method from GET to POST.

Fill in the Enter request URL text field with our API URI:

https://nameless-thunder-9740.us-east1.akkaserverless.app/com.example.CounterService/Increase

Your request should look like this now:

Click on Body.
Click on raw.

If you're not finding the above two, they're here:
Fill in the large text area with the above POST body:
```
{"counter_id": "foo", "value": 1}
```
Change Text type to JSON.
Click Send.
Profit!

Your Postman app should looking something like this. 200 OK, in green, means success!

You can rinse-and-repeat the above steps for each of the four API requests defined the protobuf file. You can also access a fully-finished Postman Collection (a set of API requests) by clicking on the below button.

Where to go next

If you're wanting to build out your first gRPC API (or just build more!), head to Akka Serverless to get a free account. Or if you want to read first, the docs are at https://developer.lightbend.com/docs/akka-serverless/quickstart/index.html.

gRPC test-and-try with Akka Serverless and Evans

Jeremy Pollock — Tue, 04 Jan 2022 03:52:33 +0000

gRPC is awesome but this post isn't going to delve too deep there. Plenty of resources out there to learn more (definitely check out https://github.com/grpc-ecosystem/awesome-grpc).

I first learned about gRPC about five years go. Since that moment in time when an engineer at PubNub introduced me to the framework, I have let the idea of gRPC simmer more in the background, especially since I was already rather steeped in REST, Open API, Swagger, and other sundries seen in the broader API space (miss you, Mashery. There seemed to be plenty of great tooling, documentation and technologies supporting "traditional" API development. Why break new ground and learn something new?

Well, when I joined my new company, Lightbend, I quickly learned that gRPC was a big part of my future! We use that framework as a big part of our new Platform-as-a-Service (PaaS), Akka Serverless, that makes it as easy to build a real-time data application as it is to build a web page (well, a simple HTML one!). Now you might be thinking, "well, gRPC isn't so easy..." but you'd be wrong, or at least ripe for disruption. It takes a bit of learning, for sure, but there are plenty of great resources to help you out. And I'm going to dig into one of these in this post. A tool that makes it easy to try out those wonderful gRPC services that I know you're going to build!

Using Evans to try out a gRPC service

Given my background in the API management space, and with not so distant memories of Mashery, Apigee and 3scale duking it out with their various flavors of API test consoles (kudos to Apigee for using their console to get ahead in the market!), one would think that I'm a sucker for UI. And yes, I do like the modern take on API test-and-try consoles; Postman being quite a great example. I was excited about their beta announcement for gRPC support! But as I dove into that tool and the other quite excellent API test-and-try tool, Kong's Insomnia, I realized that a critical feature - at least in my opinion - that was missing was server reflection, which allows for dynamic clients to be used; no necessary gathering and loading of various protobuf files. For more simplistic, single proto file implementations, not a problem. But with Akka Serverless, one can start to create rather elaborate gRPC based APIs and microservices, including those that span over event brokers. The protobuf files and structure become more sophisticated, with 3rd party library imports and the like. You can gather up all the protos and sometimes that is indeed needed. But for quick testing, especially by those who aren't gRPC experts, server reflection makes it much easier to get going.

And who am I kidding? I'm a CLI-type person. Which is why I was super excited to stumble about Evans. Within minutes, I had gone from installation to trying out TLS-secured APIs and microservices running in the cloud on Lightbend's new serverless offering.

So let's try it out!

First, Install Evans!

Following the instructions from the Github repo, install the great gRPC CLI tool, Evans. For MacOS users, it is as simple as:

$ brew tap ktr0731/evans
$ brew install evans

Now take it out for a spin!

For a terminal window, type in the following:

$ evans --tls --host nameless-thunder-9740.us-east1.akkaserverless.app  --port 443 -r repl

The above command will create an Evans CLI session, connected to a demo gRPC service running in Lightbend's Akka Serverless Platform-as-a-Service (PaaS). The demo service in question is a simple counter API by which we can increment and decrement arbitrary counters. But don't take my word for it...

First, get a list of packages available in the gRPC service, and then select the com.example one. This is setting the context for further CLI commands.

nameless-thunder-9740.us-east1.akkaserverless.app:443> show package
+-------------------------+
|         PACKAGE         |
+-------------------------+
| com.example             |
| com.example.actions     |
| grpc.reflection.v1alpha |
+-------------------------+
nameless-thunder-9740.us-east1.akkaserverless.app:443> package com.example

Now, get the list of services exposed in the package. And select the CounterService service; further defining the context, like selecting the package above.

com.example@nameless-thunder-9740.us-east1.akkaserverless.app:443> show service
+----------------+-------------------+---------------+----------------+
|    SERVICE     |        RPC        | REQUEST TYPE  | RESPONSE TYPE  |
+----------------+-------------------+---------------+----------------+
| CounterService | Increase          | IncreaseValue | Empty          |
| CounterService | Decrease          | DecreaseValue | Empty          |
| CounterService | Reset             | ResetValue    | Empty          |
| CounterService | GetCurrentCounter | GetCounter    | CurrentCounter |
+----------------+-------------------+---------------+----------------+
com.example@nameless-thunder-9740.us-east1.akkaserverless.app:443> service CounterService

Make an API call, to get the current state of the foo counter. The Evans CLI will prompt you for the counter_id.

com.example.CounterService@nameless-thunder-9740.us-east1.akkaserverless.app:443> call GetCurrentCounter
counter_id (TYPE_STRING) => foo
{
  "value": 23
}

Increase the value of the counter by some amount, say 10. Again, the Evans CLI will ask you for the counter_id and the value by which the counter should be increased.

com.example.CounterService@nameless-thunder-9740.us-east1.akkaserverless.app:443> call Increase
counter_id (TYPE_STRING) => foo
value (TYPE_INT32) => 10
{}

Finally, make another request to see the updated counter value.

com.example.CounterService@nameless-thunder-9740.us-east1.akkaserverless.app:443> call GetCurrentCounter

counter_id (TYPE_STRING) => foo

{

  "value": 33

}

Where to go next

Definitely check out Awesome gRPC for tons of great resources, even beyond the tooling like the Evans CLI. Also, if you want to start building gRPC services in a streamlined fashion, without spinning up servers and figuring out deploys and operations, then you should sign-up for an account at Akka Serverless.

Build Real-time Presence For Chat, Social and other Virtual Events

Jeremy Pollock — Fri, 10 Dec 2021 22:57:21 +0000

Thankfully, Turkey Day this year was a success, with time well spent with family and lots of drink and delicious food, although I was not able to convince my sister, responsible for food prep this year, to use my Thanksgiving Turkey of a post (hey!) in her cooking process. On this year's Thanksgiving Day, knowing that I had a day full of eating and drinking ahead of me, I took part in an annual tradition of mine: the Peloton Turkey Burn. A great way to burn calories but also to reflect on what it means to be real-time, within the context of a virtual social event with 10s of 1000s of users joining, sharing data and other wise participating in a shared experience. What technology enables this? How should it all work to yield the best experience? And when can I start eating? All questions that flowed through my mind during this year's event.

NOTE

If you want to skip to code, go [[Add Scalable Real-time Presence with Akka Serverless, Scala and Kafka#Presence Requirements|here]]. Also, you can view and then clone the entire project containing the three services, created as part of the article, here (git clone git@github.com:jpollock/akka-serverless-presence-application.git in a directory on your machine, if you want to cut to the chase!).

Each year, tens of thousands of eager-to-burn-calories-in-advance-of-massive-food-intake riders hop on to their Pelotons and jointly, live, take part in a special ride, mostly in the comfort of their homes and apartments. Normally, and you can see this anecdotally just by participating in their live rides, you might find tens to hundreds of riders taking part in these real-time, live events. Scaling architectures and infrastructure to support these ups and downs - same solution for the live rides with tens of riders as for the Turkey Burns and other major events Peloton hosts throughout the year - is tough. With Peloton, of course the main feature is the live feed; I can't ride the bike if I don't have that instructor telling me what to do! But much of the success of their platform is based on the social features in the bike's application: from giving high-fives to fellow riders, to tracking position in the race - ahem, I mean 'ride' - to enabling one to filter out those pesky young riders so that this 50+ male from SF can feel good about his relative place in the world, the Peloton experience relies heavily on the idea of "presence": who is on-line, what are they all about, i.e. their profile, and what's happening relative to them, in the context of the event, in the case of Peloton. This is not much different from the presence of chat applications, on-line social events and even platforms like LinkedIn where they use #akka to deliver their real-time presence platform. Unless things have changed, Peloton leverages some real-time capabilities from PubNub and as it happens, I was the Product Manager there during the development of a cool new feature that we built, based on Peloton's requirements: Presence Search.

Each approach can yield success, either through a roll-your-own-solution using a technology like Lightbend's Akka or an As-A-Service offering like PubNub. It certainly depends on your needs; developers are often attracted to off-the-shelf offerings because of some of the inherent scaling challenges with presence, especially in those situations where usage is highly elastic. Dream11, another Lightbend customer and user of Akka, is another great example of this: they successfully use Akka to handle, efficiently, the real-time sensitive joining of users into live events. To quote that case study, linked above:

Each contest can have as few as two participants to upwards of tens of millions that can join up until the real-world event begins.

In either case, DIY and hosted API come with pros and cons. DIY gives you maximum control but comes with potential costs of having to think that much more deeply around architecture as well as supporting operationally highly elastic use cases. Hosted APIs are great but can sometimes cut down on flexibility of solution and comes with inherent availability risks of the the 3rd party vendor's services.

What if you could get a little bit of the best of both worlds? To build a Peloton, a Dream11, a LinkedIn-esque real-time presence platform, to drive your use case, whether chat, on-line events, social applications and event IoT scenarios, all with the flexibility, robustness and scalability of a DIY framework like Akka, with the added value of not having to operate the services themselves?

There is an answer! Akka Serverless!

Presence Requirements

Before diggiing into code, let's just spend a moment on some basic requirements. I won't be fully replicating what LinkedIn, Dream11 and Peloton have done but it won't be too far off either.

Need	Feature to Build	Perf/scale requirements
Tracking on-line status of users	On-line/off-line tracker with heartbeats used to update a user's status	Possibly break out heartbeats from state management
Provide ability to augment user profile with real-time data	Flexible data model with real-time data integration support	Possibly keep separate from heartbeats but with state management
Provide queryability to the user data	Search for state management, including on-line status and profile attributes	Keep separate from heartbeats and possibly from core state management, i.e. #CQRS

For our solution, from the above, we'll create three separate services, using Akka Serverless for API logic and Kafka for event distribution. It could be that, in practice, we might not need them to developed as three seperate services but I do so such that the reader can see the art of possible with Akka Serverless. And if I were really building such a product or application, I would choose to do it this way, given my past experiences, product managing a very large virtual events API platform (PubNub). Through those experiences, and learnings from Lightbend as well, I can see:

Spiky usage, with large scale on-line events - think on-line concert, perhaps - driving very sudden increases in heartbeats and queries but perhaps similar increases for actual user profile data changes;
Varied audience sizes - think normal Peloton class versus that Turkey Burn with 50K+ riders - along with varied usage scenarios, such that provisioning capacity becomes difficult to do;
Longer durations between heartbeating but with higher usage of querying for IoT use cases.

To build this application that provides optimal performance and flexible scalability, I do so, with the following diagram leading the way:

)

This is going to give us the flexibility of scaling each component as needed and promote easier feature changes. I choose multiple languages (Java, Scala, Python) simply to show off the polyglot nature of Akka Serverless, which can help different teams build different pieces of the solution, in the language of their choice. Also, for a great developer experience, I use the Kafka-compatible Redpanda from Vectorized as the inter-service messaging solution.

Let's start!

First, let's clone the repository, assuming that you want to look at the full files and even play with the service itself. Run the following in a directory of your choosing.

git clone git@github.com:jpollock/akka-serverless-presence-application.git

Heartbeats Service (Java)

Every time that I start an Akka Serverless project, using either Java or Scala, I use the project template. And thus it is true for the Heartbeats one. Definitely check-out that link for when you want to start your own development work. But for now, we're not going to step through the process. Not quite "in media res", the moment of API data domain definition and API specification is where I will pick up.

Domain Data

In src/main/proto/com/example/demo/presence/domain/user_presence_domain.proto, I have defined the data schema for the heartbeat domain, using protocol buffers.

syntax = "proto3";

package com.example.demo.presence.domain;

import "akkaserverless/annotations.proto";

option (akkaserverless.file).replicated_entity = {
    name: "UserPresenceHeartbeat"
    entity_type: "user_presence_heartbeat"
    replicated_register: { 
        value: "Heartbeat" 
    }
};

message Heartbeat {
    string user_id = 1;
    string device_id = 2;
    bool is_online = 3;
    Profile profile = 4;
    bool is_modified = 5;
}

message Profile {
    string attr1  = 1;
    string attr2 = 2;
}

Most of the above is pretty straightforward. While how your domain is being modeled is most important, in order to use Akka Serverless, I do need to pick the state model. This is set in this snippet of protobuf:

option (akkaserverless.file).replicated_entity = {
    name: "UserPresenceHeartbeat"
    entity_type: "user_presence_heartbeat"
    replicated_register: { 
        value: "Heartbeat" 
    }
};

For presence heartbeats, I have decided to use Akka Serverless's Replicated Entity state model. There are several Conflict-Free Replicated Data Types (CRDTs) to pick from; in this case, I want to store just a value for each user/device combination. That's why I pick the Replicated Register CRDT.

name: denotes the base name for the Replicated Entity, the java code-generation process will create initial sources UserPresenceHeartbeat, UserPresenceHeartbeatTest and UserPresenceHeartbeatIntegrationTest. Once these files exist, they are not overwritten, so you can freely add logic to them.
entity_type: denotes a unique identifier of the "state storage". The entity name may be changed even after data has been created, the entity_type can’t.
replicated_register: denotes the CRDT to be used.
value: denotes the reference message definition, i.e. a pointer to another object defined in the protobuf.

API Specification

In src/main/proto/com/example/demo/presence/user_presence_api.proto, I have defined the facade, the specification of what API requests I will support and what their inputs and outputs are.

// This is the public API offered by your entity.
syntax = "proto3";

import "google/protobuf/empty.proto";
import "akkaserverless/annotations.proto";
import "google/api/annotations.proto";

import "com/example/demo/presence/domain/user_presence_domain.proto";

package com.example.demo.presence;

message PresenceHeartbeatCommand {
    string user_id = 1 [(akkaserverless.field).entity_key = true];
    string device_id = 2 [(akkaserverless.field).entity_key = true];
    bool is_online = 3;
    domain.Profile profile = 4;
}

service PresenceHeartbeatCommandService {
    option (akkaserverless.service) = {
        type : SERVICE_TYPE_ENTITY
        component : "com.example.demo.presence.domain.UserPresenceHeartbeat"
    };

    rpc Heartbeat(PresenceHeartbeatCommand) returns (domain.Heartbeat) {
    }
}

service PresenceHeartbeatEventService {
    option (akkaserverless.service) = {
        type : SERVICE_TYPE_ACTION
    };

    rpc HeartbeatEvent(PresenceHeartbeatCommand) returns (domain.Heartbeat) {
        option (google.api.http) = {
            put: "/users/{user_id}/devices/{device_id}/heartbeat"
            body: "*"
        }; 
    }
    rpc HeartbeatEventToKafka(domain.Heartbeat) returns (domain.Heartbeat) {
        option (akkaserverless.method).eventing.out = {
            topic: "heartbeats"
        };    
    }
}

There are many interesting aspects to this API specification.

[(akkaserverless.field).entity_key = true] in the PresenceHeartbeatCommand message definition are special Akka Serverless annotations; this one lets the run-time connect your incoming API request to specific entity instances (not quite DB rows but moe like uniquely addressable objects running in memory, with a backing DB behind it), e.g. user_is=jeremy and device_id=iphone would route incoming requests to the Akka Serverless entity with those two paticular attribute/value combinations. And given the entity_key serves as a unique identifier, we know that there is only one instance of the entities.
There's another Akka Serverless annotation and that is the one that starts with option (akkaserverless.service) = { in the above (first line below the beginning of the PresenceHeartbeatCommandService service definition). This is used by Akka Serverless code-gen processes so that plenty of boilerplate code can automatically be generated for you. In this case, we're telling the code-gen process what type of component code to write, e.g. Action or Entity or View.
The last Akka Serverless annotation in this particular protobuf is the one associated with the rpc HeartbeatEventToKafka API call definition. In this case, we're letting Akka Serverless that this API is going to produce events - data - on to a message broker, with each event a message, sent to either Google Pubsub or Kafka.
For supporting incoming web requests, ala REST, Google has some annotations itself; this is the google.api.http entry, towards the bottoms of the protobuf, associated with the rpc HeartbeatEvent API call definition.

API Implementation

In the above protbuf, you may have noticed that we have two service definitions: one for the CRDT entity that will store the actual heartbeat data and one for the action that will ingest data events over HTTP/S. Why not just have the entity itself respond directly to HTTP/S requests? Because entities cannot emit events to external message brokers. But we can use the Actions as Controllers pattern.

In src/main/java/com/example/demo/presence/PresenceHeartbeatEventServiceAction.java, we see our logic:

/** An action. */
public class PresenceHeartbeatEventServiceAction extends AbstractPresenceHeartbeatEventServiceAction {

    public PresenceHeartbeatEventServiceAction(ActionCreationContext creationContext) {}

    /** Handler for "HeartbeatEvent". */
    @Override
    public Effect<UserPresenceDomain.Heartbeat> heartbeatEvent(UserPresenceApi.PresenceHeartbeatCommand presenceHeartbeatCommand) {
        CompletionStage<UserPresenceDomain.Heartbeat> heartbeatDone =
            components().userPresenceHeartbeat().heartbeat(presenceHeartbeatCommand)
                .execute();            

        CompletionStage<Effect<UserPresenceDomain.Heartbeat>> effect = heartbeatDone.thenApply(heartbeat -> {
        System.out.println("HeartbeatEvent: " + heartbeat);

        if (heartbeat.getIsModified()) {
            DeferredCall<UserPresenceDomain.Heartbeat, UserPresenceDomain.Heartbeat> call =
                components().presenceHeartbeatEventServiceAction().heartbeatEventToKafka(heartbeat); 
            return effects().forward(call);
        } else {
            return effects().reply(heartbeat);
        }

        });
        System.out.println("Heartbeat Effect: " + effect);
        return effects().asyncEffect(effect);

    }

    /** Handler for "HeartbeatEventToKafka". */
    @Override
    public Effect<UserPresenceDomain.Heartbeat> heartbeatEventToKafka(UserPresenceDomain.Heartbeat heartbeat) {
        System.out.println("HeartbeatEventToKafka: " + heartbeat);
        return effects().reply(heartbeat);

    }
}

The method heartbeatEvent is the logic that will be executed whenever an API request is made to the endpoint specified in the protbuf file. This is what we call a command handler, i.e. the business logic that you want executed for given requests. In this particular handler, I'm using the service composability capabilities in Akka Serverless to make another request - internal to the product - to the actual entity service that is managing my heartbeat CRDT data. I access this through the service's components, with userPresenceHeartbeat matching the name of the component as defined in the service's protobuf definition. The method that I call, heartbeat, is codified in src/main/java/com/example/demo/presence/domain/UserPresenceHeartbeat.java. That code will be executed and once complete, returned to the action, which then determines if it should immediately reply or forward the logic flow to another action handler, heartbeatEventToKafka, which ultimately is the control point for sending events to Google Pubsub or Kafka, given the out to a topic definition in the protobuf file. We don't want to unnecessarily send unchanged data events so we use the logic of a simple if statement, e.g. if (heartbeat.getIsModified()) { to determine whether to forward or not. If forwarded, the data will be sent to the target message broker topic.

We know now how the events are ingested and egressed out of the service and the logic in between. Well, part of the logic, that is; let's dig a bit into the CRDT entity code.

In src/main/java/com/example/demo/presence/domain/UserPresenceHeartbeat.java, we see our entity logic:

/** A replicated entity. */
public class UserPresenceHeartbeat extends AbstractUserPresenceHeartbeat {
    @SuppressWarnings("unused")
    private final String entityId;

    public UserPresenceHeartbeat(ReplicatedEntityContext context) {
        this.entityId = context.entityId();
    }

    @Override
    public UserPresenceDomain.Heartbeat emptyValue() {
        return UserPresenceDomain.Heartbeat.getDefaultInstance();
    }

    @Override
    public Effect<UserPresenceDomain.Heartbeat> heartbeat(ReplicatedRegister<UserPresenceDomain.Heartbeat> currentData, UserPresenceApi.PresenceHeartbeatCommand presenceHeartbeatCommand) {
        System.out.println("UserPresenceHeartbeat.heartbeat.state: " + currentData);
        UserPresenceDomain.Heartbeat currentHeartbeat = currentData.get(); 

        boolean isModified = true;
        if (currentHeartbeat.getIsOnline() == presenceHeartbeatCommand.getIsOnline()) {
            isModified = false;
        } 
        UserPresenceDomain.Heartbeat newValue = UserPresenceDomain.Heartbeat.newBuilder()
            .setUserId(presenceHeartbeatCommand.getUserId())
            .setDeviceId(presenceHeartbeatCommand.getDeviceId())
            .setIsOnline(presenceHeartbeatCommand.getIsOnline())
            .setProfile(presenceHeartbeatCommand.getProfile())
            .setIsModified(isModified)
            .build();
        System.out.println("UserPresenceHeartbeat.heartbeat: " + newValue);
        return effects()
            .update(currentData.set(newValue)) 
            .thenReply(newValue);  

    }
}

Holy cow! System.out.println??? Well, check my bio. I blame being a Product Manager! Anyway, the critical method, heartbeat, is not doing anything terribly sophistiated: we look at the current state of the entity - that's automatically handed to us as part of the Akka Serverless proxy - and, if not changed, do nothing, else, update the state. We use the isModified property to tell the calling action whether or not to forward the heartbeat event to the message broker. We only emit events for state changes; that allows us to scale each service differently, e.g. if 1000s of users are on-line and firing heart beat events every 5 seconds, we don't push that load to the other services unless actual changes are occurring.

I didn't step through all of the code, line by line, but it should be faily self-apparent, especially given the small scope of actual logic. Let's fire up the service and quickly test it out.

Trying the Java Heartbeats Service

Before you start, make sure Docker is running!

In a terminal window, get into the root of the project directory, cloned above, e.g. cd presence-heartbeats-java;
In the same terminal window, start the partially cooked Turkey API: PORT=8081 mvn compile exec:exec;
In another terminal window, in the root of the project directory: docker-compose up.

User State Service (Scala)

Domain Data

In src/main/proto/com/example/demo/presence/domain/user_presence_domain.proto, I have defined the data schema for the user_presence domain, using protocol buffers. This is the actual user state, including profile attributes that might be specific to an application and/or use case.

syntax = "proto3";

package com.example.demo.presence.domain;

import "akkaserverless/annotations.proto";

option (akkaserverless.file).value_entity = {
    name: "UserPresence"
    entity_type: "user_presence"
    state: "UserPresenceState"
};

message UserPresenceState {
    string user_id = 1;
    string device_id = 2;
    bool is_online = 3;
    Profile profile = 4;
}

message Heartbeat {
    string user_id = 1 [(akkaserverless.field).entity_key = true];
    string device_id = 2 [(akkaserverless.field).entity_key = true];
    bool is_online = 3;
    Profile profile = 4;
}

message Profile {
    string attr1  = 1;
    string attr2 = 2;
}

Like our heartbeats service, since we're designing our domain data model, we need to pick our state model. In this case, I've gone with Value Entity, as seen by option (akkaserverless.file).value_entity above. Value entities are essentialy a Key-Value store.

name: denotes the base name for the Value Entity, the scala code-generation process will create initial sources UserPresence, UserPresenceTest and UserPresenceIntegrationTest. Once these files exist, they are not overwritten, so you can freely add logic to them.
entity_type: denotes a unique identifier of the "state storage". The entity name may be changed even after data has been created, the entity_type can’t.
state: denotes the protobuf message representing the Value Entity’s state which is kept by Akka Serverless.

API Specification

In src/main/proto/com/example/demo/presence/user_presence_api.proto, I have defined the facade, the specification of what API requests I will support and what their inputs and outputs are.

// This is the public API offered by your entity.
syntax = "proto3";

import "google/protobuf/empty.proto";
import "akkaserverless/annotations.proto";
import "google/api/annotations.proto";

import "com/example/demo/presence/domain/user_presence_domain.proto";

package com.example.demo.presence;


message GetUserPresenceCommand {
    string user_id = 1 [(akkaserverless.field).entity_key = true];
    string device_id = 2 [(akkaserverless.field).entity_key = true];

}

service PresenceStateService {
    option (akkaserverless.service) = {
        type : SERVICE_TYPE_ENTITY
        component : "com.example.demo.presence.domain.UserPresence"
    };

    rpc TogglePresence(domain.Heartbeat) returns (domain.UserPresenceState) {}

    rpc GetCurrentUserPresence(GetUserPresenceCommand) returns (domain.UserPresenceState) {
        option (google.api.http) = {
            get: "/users/{user_id}/devices/{device_id}"
        };      
    }

}

service PresenceHeartbeatEventService {
    option (akkaserverless.service) = {
        type : SERVICE_TYPE_ACTION
    };


    rpc HeartbeatEvent(domain.Heartbeat) returns (domain.UserPresenceState) {
        option (akkaserverless.method).eventing.in = {
            topic: "heartbeats"
        }; 
        option (akkaserverless.method).eventing.out = {
            topic: "users"
        }; 

    }
}

Given our description of the Java Heartbeats service's protobuf, the above should make sense. Everything is basically an analog of that other protobuf, from the Akka Serverless' annotations for conntecting to entity system models, to the mapping of events to pubsub/kafka topics.

API Implementation

You're looking for my println statements, right? Not in my scala, I say! Anyway, similar to the Heartbeats service, we define two APIs in our protobuf: one for the Key-Value entity that will store the actual user data and one for the action that will ingest and emit data events via messaging (Pubsub or Kafka). Remember: this is the Actions as Controllers pattern.

In src/main/java/com/example/demo/presence/PresenceHeartbeatEventServiceAction.scala, we see our logic:

/** An action. */
class PresenceHeartbeatEventServiceAction(creationContext: ActionCreationContext) extends AbstractPresenceHeartbeatEventServiceAction {

    /** Handler for "HeartbeatEvent". */
    override def heartbeatEvent(heartbeat: Heartbeat): Action.Effect[UserPresenceState] = {

        val heartbeatReply: Future[UserPresenceState] =
        for { 
            userPresenceState <- components.userPresence.togglePresence(heartbeat).execute()
        } yield userPresenceState

        effects.asyncReply(heartbeatReply)
    }
}

The method heartbeatEvent is the logic that will be executed whenever a message, of type Heartbeat, is received on the message broker topic, heartbeats. Like our Java code, this is a command handler as well, i.e. the business logic that you want executed for given requests. In this particular handler, I'm using the service composability capabilities in Akka Serverless to make another request - internal to the product - to the actual entity service that is managing my user state Value Entity (Key-Value) data. I access this through the service's components, with userPresence matching the name of the component as defined in the service's protobuf definition. The method that I call, heartbeat, is codified in src/main/java/com/example/demo/presence/domain/UserPresence.scala. That code will be executed and once complete, returned to the action, which will we return it to the calling application, and, given the out to a topic definition in the protobuf file, also to either Google Pubsub or Kafka.

Let's dig a bit into the Key-Value entity code.

In src/main/java/com/example/demo/presence/domain/UserPresence.scala, we see our entity logic:

/** A value entity. */
class UserPresence(context: ValueEntityContext) extends AbstractUserPresence {
override def emptyState: UserPresenceState = UserPresenceState()

override def togglePresence(currentState: UserPresenceState, heartbeat: Heartbeat): ValueEntity.Effect[UserPresenceState] = {
    val newState = currentState.copy(userId = heartbeat.userId, 
    deviceId = heartbeat.deviceId, 
    isOnline = heartbeat.isOnline,
    profile = heartbeat.profile)

    effects
        .updateState(newState) 
        .thenReply(newState) 
}

override def getCurrentUserPresence(currentState: UserPresenceState, getUserPresenceCommand: presence.GetUserPresenceCommand): ValueEntity.Effect[UserPresenceState] = 
    effects.reply(currentState)
}

No println! This file contains two methods, one to enable fetching user state through a GET call, i.e. the getCurrentUserPresence method, which simply returns the current state of the user. The other method, togglePresence, takes the event data from the message broker and transforms the Heartbeat into a UserPresenceState, signaling to Akka Serverless to save the update, via the updateState call and then return the new data back to the action, which in turn will emit that update out as an event published to the message broker topic, users.

Pretty simple, huh? Let's try this out!

Trying the Scala User State Service

Before you start, make sure Docker is running!

In a terminal window, get into the root of the project directory, cloned above, e.g. cd presence-user-state-scala;
In the same terminal window, start the partially cooked Turkey API: sbt run;
In another terminal window, in the root of the project directory: docker-compose up.

Querying Service (Python)

So far we've walked through the code for our Heartbeats service, which is able to be scaled up and down automatically in Akka Serverless depending on load and use case, handling the ingestion of potentially highly variable user on-line/off-line events. This was in Java. For actual storage of the users and their current state, we built our User State service, which uses an underling Key-Value state model, i.e. Value Entity, for storage and quick access to a user's current data. Both services have events flowing between using the message broker integration in Akka Serverless, whether Google Pubsub or Kafka based.

The last piece of our architectural puzzle is our query service. This enables a #CQRS implementation, with querying in a separate service and as such can be optimized for performance separately from the other pieces of our overall presence application. I built mine using the community Python SDK. Whenever I start a new Python service, I typically go here. There's no code-gen (yet) for Python SDK.

The feature that we are using for the query service is called Views. They enable you to query over the state. From https://developer.lightbend.com/docs/akka-serverless/reference/glossary.html#view:

A View provides a way to retrieve state from multiple Entities based on a query. You can query non-key data items. You can create views from Value Entity state, Event Sourced Entity events, and by subscribing to topics.

We'll use the "subscribing to topics" in our approach.

Domain Data

The Python SDK works a bit different that Java or Scala. A single file protobuf is best and this is what I have done for now. The domain data I am dealing with is the same as defined in our Scala SDK, UserPresenceState. That is what is being emitted from the User State service and what the view feature expects, in order to build up the underlying query implementation. The below should be straightforward, with nothing new; remember that the [(akkaserverless.field).entity_key = true] tell Akka Serverless what the entity instance identifier fields are.

message UserPresenceState {
    string user_id = 1 [(akkaserverless.field).entity_key = true];
    string device_id = 2 [(akkaserverless.field).entity_key = true];
    bool is_online = 3;
    Profile profile = 4;
}

message Profile {
    string attr1  = 1;
    string attr2 = 2;
}

API Specification

message UsersResponse {
    repeated domain.UserPresenceState results = 1; 
}

service PresenceQueryApi {
    rpc UpdateView(UserPresenceState) returns (UserPresenceState) {
        option (akkaserverless.method).eventing = {

        in: {
            consumer_group: "python-consumer-group"
            topic: "users"
            }
        };
        option (akkaserverless.method).view.update = {
            table: "users"
        };
    }

    rpc GetUsers(google.protobuf.Empty) returns (UsersResponse) {
        option (akkaserverless.method).view.query = {
            query: "SELECT * AS results FROM users"
        };
        option (google.api.http) = {
            get: "/users"
        };  
    }  
}

API Implementation

This is when it gets really complicated! Well, there really isn't an implementation. Akka Serverless does need to have some boilerplate code in order to do its thing but that's it. In api_impl.py we find:

# imports fom Akka Serverless SDK
from akkaserverless.view import View

# imports fom Python generated code
from api_spec_pb2 import (_PRESENCEQUERYAPI, DESCRIPTOR as FILE_DESCRIPTOR)

view = View(_PRESENCEQUERYAPI,[FILE_DESCRIPTOR])

All we're doing is writing some boilerplate to wire everything up, connecting the view to the API definition. We see this in the inclusion of the _PRESENCEQUERYAPI import and passing that into the View Akka Serverless Python object.

Trying the Python Querying Service

Before you start, make sure Docker is running!

In a terminal window, get into the root of the project directory, cloned above, and enter the python folder e.g. cd presence-querying-python;
You will want to create a Virtual Environment: python3 -m venv akkasls_env (akkasls_env can be named whatever you want and located anywhere though);
Initiate the virtual environment: source akkasls_env/bin/activate;
Install the needed Python libraries: pip install -r requirements.txt;
In the same terminal window (or another with the virtual environment activated), start the Python service: PORT=8082 USER_FUNCTION_PORT=$PORT start.sh.
In another terminal window, in the same directory, with the virtual environment sourced, simulate an event using kafka_message_generator.py.

Deploying and Testing in Akka Serverless

We have three separate services as part of our application. This makes testing all locally a bit challenging, since your computer might not have all of the resources to run everything (start_all.sh in the root of the project is there if you want though). So rather than do that, let's just deploy to Akka Serverless.

Install the Akka Serverless CLI;
Once installed, you can sign-up for a new account via akkasls auth signup (or visit the Console Sign-up page;
Login to your account via the CLI: akkasls auth login;
In a terminal window, in the root directory of the project, set environmental variables: export DOCKER_REGISTRY= export DOCKER_PASS=
Configure your Kakfa broker per the [documentation](https://developer.lightbend.com/docs/akka-serverless/projects/message-brokers.html#_confluent_cloud. Do not pick the Python client, since the actual Kafka integration is running in the Akka Serverless proxy. You can also configure the broker to use other Kafka platforms; we love Redpanda.
Per the same documentation, create the users and heartbeats topics in Confluent Cloud or Redpanda.
In the same terminal window, run the command deploy.sh.
Make some API calls!

-- send a heartbeat (java service)
curl -XPUT -H "Content-Type: application/json" "https://$(akkasls services get presence-heartbeats-java -o json | jq -r '.contourroutes[0].spec.host')/users/myuser/devices/mydevice/heartbeat" -d '{"is_online": true, "profile": {"attr1": "test", "attr2": "test"}}'

-- get the user's state (scala service, updated via Kafka events)
curl -XGET "https://$(akkasls services get presence-user-state-scala -o json | jq -r '.contourroutes[0].spec.host')/users/myuser/devices/mydevice"

-- query the list of users (python service, updated via Kafka events)
curl -XGET "https://$(akkasls services get presence-querying-python -o json | jq -r '.contourroutes[0].spec.host')/users"

Conclusion

Hopefully you now have a better idea on how some of the moving parts work in Akka Serverless and how you can build event driven APIs and services, including those that can be integrated with Kafka. We also leverage much of the new service composability features in Akka Serverless, to make a rather sophisticated application without a whole lot of setup.

And don't forget: get your free Akka Serverless account here.

Distributed Real-Time Turkey Cooking With Scala, Akka Serverless and Kafka

Jeremy Pollock — Tue, 23 Nov 2021 19:06:48 +0000

Right now, my mind turns toward the wonderful feast awaiting me this coming Thanksgiving Thursday. But before I dig into the turkey, stuffing and pumpkin pie, I thought that I would whet my appetite with an antipasto of Scala, Akka Serverless and Kafka.

Thanksgiving is a big holiday here in the United States. It is a day when family and friends gather together to celebrate life and give thanks to the bounty that we enjoy. Also, we like to eat and drink on this day. A lot. And the foods we tend to enjoy are most often long observed traditions such as stuffing, pumpkin pies, cranberry sauce, mashed potatoes with the gravy made from the juices and nicely browned bits of the piece de resistance, the Thanksgiving Turkey.

Gluttonous consumption seems rather distasteful these days so I thought that it would be great to see how we could leverage technology to make sure that the food that we do being to the table is done so most efficiently and with maximum great taste. What better tools to bring into the picture than Akka - in the form of the new serverless offering - and Kafka?

There is plenty of content and data out there supporting the assertion that Akka is a highly efficient distributed data and streaming platform. These are some good articles and posts: Towards a Reliable Device Management Platform, Streaming a Million Likes/Second: Real-Time Interactions on Live Video, Actors for Akka Stream Throughput. But let's talk turkey!

Wait. Is this another blog post about the perfect recipe or most effective cooking method for that large bird you've procured from the local supermarket? It is not. In fact, this blog post is intended really to let us play a bit with an interesting new developer platform product: Akka Serverless. And to do so with a fun, funny and utterly useless use case: how to enable a distributed team of cooks, experts, and industry, to create the perfect turkey!

In order to not stuff ourselves too much, we'll keep our focus tight:

Standup a general turkey management API, using the event sourcing capabilities of Akka Serverless;
Integrate Kafka into the API, simulating an industry-hosted Machine Learning-based recommender system.

At the end of this article, you will have prepared yourself fully for a real Thanksgiving Feast!

Let's get going!

For this exercise, we will start with a partially cooked project. Go ahead and get your environment setup by cloning: git clone git@github.com:jpollock/turkey_tracker_scala.git. The main brach is our starter, the partially cooked turkey, if you will. There are other branches that you can pull to get further along the cooking process. Oops. I mean "coding process"!

The schema_change branch has the modifications we will do in the first part of this exercise;
The kafka_integration branch has the modifications we will do in the latter part, i.e. when we hook up Kafka to Akka Serverless!

What The Turkey Expects

In order to walk through the code in this post, you're going to want the following set up in your environment:

Initial State

To kick things off, let's walk through the initial part of the exercise..

In a terminal window, get into the root of the project directory, cloned above, e.g. cd turkey_tracker_scala;
In the same terminal window, start the partially cooked Turkey API: sbt run;
In another terminal window, in the root of the project directory: docker-compose -f docker-compose-proxy.yml.

Make Some API Calls

Let's start cooking!

curl -X POST -H 'Content-Type: application/json' http://localhost:9000/com.example.TurkeyService/StartCooking -d '{"turkey_id": "myfirstturkey"}'

curl -X POST -H 'Content-Type: application/json' http://localhost:9000/com.example.TurkeyService/GetCurrentTurkey -d '{"turkey_id": "myfirstturkey"}'

The response should be:

{"inOven":true,"done":"RAW","externalTemperature":0.0}%

Let's stop cooking!

curl -X POST -H 'Content-Type: application/json' http://localhost:9000/com.example.TurkeyService/EndCooking -d '{"turkey_id": "myfirstturkey"}'

curl -X POST -H 'Content-Type: application/json' http://localhost:9000/com.example.TurkeyService/GetCurrentTurkey -d '{"turkey_id": "myfirstturkey"}'

The response should be:

{"inOven":false,"done":"RAW","externalTemperature":0.0}%

But the question remains: Was the turkey tasty?

Modifying the API to prevent salmonella poisoning

NOTE: You can git clone schema_changes to get all of the code changes below.

Before determining taste though, the more important question is Has the turkey reached an internal temperature such that salmonella bacteria have been killed? Because, you know, having folks visit the hospital on account of Thanksgiving eating is not desirable.

Let's look at the current data model for our Turkey. Akka Serverless uses Protocol Buffers as the specification format for the data and event model as well as the API interfaces. The below is the contents from the turkey_domain.proto file.

syntax = "proto3";

package com.example.domain;

import "akkaserverless/annotations.proto";

option (akkaserverless.file).event_sourced_entity = {
    name: "Turkey"
    entity_type: "turkey"
    state: "TurkeyState",
    events: ["InOven", "OutOfOven", "TemperatureChange"]
};

message TurkeyState {
    bool in_oven = 1;
    enum DoneStatus {
        RAW = 0;
        SALMONELLA = 1;
        STILL_SALMONELLA = 2;
        ALMOST_THERE = 3;
        PERFECT = 4;
    }
    DoneStatus done = 2;
    float external_temperature = 3;
}

message InOven {
    string turkey_id = 1 [(akkaserverless.field).entity_key = true];
}
message OutOfOven {
    string turkey_id = 1 [(akkaserverless.field).entity_key = true];
}

message TemperatureChange {
    string turkey_id = 1 [(akkaserverless.field).entity_key = true];
    float new_temperature = 3;
}

There is much to unpack here but for now, let's focus on the mission at hand: preventing salmonella poisoning by fully cooking our bird! In the above, we see that we're tracking the external temperature. That's the oven. And we can imagine increasing or decreasing throughout the cooking process in order to ensure optimal results. But we won't know that unless we start tracking what the USDA recommends: internal temperature.

We can do that by adding a new attribute, internal_temperature to the TurkeyState:

message TurkeyState {
    bool in_oven = 1;
    enum DoneStatus {
        RAW = 0;
        SALMONELLA = 1;
        STILL_SALMONELLA = 2;
        ALMOST_THERE = 3;
        PERFECT = 4;
    }
    DoneStatus done = 2;
    float external_temperature = 3;
    float internal_temperature = 4;
}

That takes care of the domain data that we will manage. But how to communicate to the bird - or anyone else - that the internal temperature has indeed changed? In the above proto file, we see that we already have a TemperatureChange event defined. This is used in the IncreaseOvenTemperature and DecreaseOvenTemperature API signatures in the turkey_api.proto. Let's try to re-use this approach by modifying the TemperatureChange message schema.

message TemperatureChange {
    string turkey_id = 1 [(akkaserverless.field).entity_key = true];
    enum Type {
        INTERNAL = 0;
        EXTERNAL = 1;
    }
    Type type = 2;
    float new_temperature = 3;    
}

Now we can identify the type of change and, as we will see in just a moment, store the data in the right place. In order to store data though we have to receive it. Let's move over to our API specification in the turkey_api.proto file.

// This is the public API offered by your entity.
syntax = "proto3";

import "google/protobuf/empty.proto";
import "akkaserverless/annotations.proto";
import "google/api/annotations.proto";
import "turkey_domain.proto";

package com.example;

message CookingCommand {
    string turkey_id = 1 [(akkaserverless.field).entity_key = true];
}
message TemperatureChangeCommand {
    string turkey_id = 1 [(akkaserverless.field).entity_key = true];
    float temperature_change = 2;
}

message GetTurkeyCommand {
    string turkey_id = 1 [(akkaserverless.field).entity_key = true];
}

service TurkeyService {
    option (akkaserverless.service) = {
        type : SERVICE_TYPE_ENTITY
        component : "com.example.domain.Turkey"
    };
    rpc StartCooking(CookingCommand) returns (google.protobuf.Empty);
    rpc EndCooking(CookingCommand) returns (google.protobuf.Empty);
    rpc IncreaseOvenTemperature(TemperatureChangeCommand) returns (google.protobuf.Empty);
    rpc DecreaseOvenTemperature(TemperatureChangeCommand) returns (google.protobuf.Empty);

    rpc GetCurrentTurkey(GetTurkeyCommand) returns (domain.TurkeyState);
}

We see that we have API signatures - Commands - for increasing and decreasing oven temperature. Let's support increasing and decreasing turkey temperature. We can add the following to the above protobuf file.

    rpc IncreaseTurkeyTemperature(TemperatureChangeCommand) returns (google.protobuf.Empty);
    rpc DecreaseTurkeyTemperature(TemperatureChangeCommand) returns (google.protobuf.Empty);

So far, we have modified our data schema and API specification to account for this need to prevent salmonella poisoining. Now we have to add the appropriate logic in our Scala code to account for this change. Let's look at our src/main/scala/domain/Turkey.scala file.

As mentioned earlier in this post, we are using Event Sourcing as the state model for Akka Serverless. For the file that we're inspecting now - the logic behind getting data in and out of Akka - that means that we need to have code for handling both commands and events. We see both for our oven temperature changes; snippets for the IncreaseOvenTemperature flow below.

override def increaseOvenTemperature(currentState: TurkeyState, temperatureChangeCommand: example.TemperatureChangeCommand): EventSourcedEntity.Effect[Empty] = {
    effects
    .emitEvent(TemperatureChange(turkeyId=temperatureChangeCommand.turkeyId, newTemperature=(currentState.externalTemperature + temperatureChangeCommand.temperatureChange))) 
    .thenReply(_ => Empty.defaultInstance) 
}

override def temperatureChange(currentState: TurkeyState, temperatureChange: TemperatureChange): TurkeyState = {
    currentState.copy(externalTemperature = temperatureChange.newTemperature)
}

When an API request is issued to Akka Serverless, to increase the temperature of the oven, that request is handled by the command handler above, which creates the TemperatureChange event and emits it. That event is subsequently, and asynchronously, handled by the event handler, which updates the actual state of the turkey with the new external_temperature. We need to make some modifications, to account for the type attribute of the TemperatureChange message schema, so we can reuse that, distinguishing between internal and external updates.

Let's first update the existing code, adding in the type attribute:

override def increaseOvenTemperature(currentState: TurkeyState, temperatureChangeCommand: example.TemperatureChangeCommand): EventSourcedEntity.Effect[Empty] = {
    effects
    .emitEvent(TemperatureChange(turkeyId=temperatureChangeCommand.turkeyId, changeType=TemperatureChange.Type.EXTERNAL, newTemperature=(currentState.externalTemperature + temperatureChangeCommand.temperatureChange))) 
    .thenReply(_ => Empty.defaultInstance) 
}

override def decreaseOvenTemperature(currentState: TurkeyState, temperatureChangeCommand: example.TemperatureChangeCommand): EventSourcedEntity.Effect[Empty]= {
    effects
    .emitEvent(TemperatureChange(turkeyId=temperatureChangeCommand.turkeyId, changeType=TemperatureChange.Type.EXTERNAL, newTemperature=(currentState.externalTemperature - temperatureChangeCommand.temperatureChange))) 
    .thenReply(_ => Empty.defaultInstance) 
}

Let's add the new command handlers for the increasing and decreasing of turkey temperature.

override def increaseTurkeyTemperature(currentState: TurkeyState, temperatureChangeCommand: example.TemperatureChangeCommand): EventSourcedEntity.Effect[Empty] = {
    effects
    .emitEvent(TemperatureChange(turkeyId=temperatureChangeCommand.turkeyId, changeType=TemperatureChange.Type.INTERNAL, newTemperature=(currentState.externalTemperature + temperatureChangeCommand.temperatureChange))) 
    .thenReply(_ => Empty.defaultInstance) 
}

override def decreaseTurkeyTemperature(currentState: TurkeyState, temperatureChangeCommand: example.TemperatureChangeCommand): EventSourcedEntity.Effect[Empty]= {
    effects
    .emitEvent(TemperatureChange(turkeyId=temperatureChangeCommand.turkeyId, changeType=TemperatureChange.Type.INTERNAL, newTemperature=(currentState.externalTemperature - temperatureChangeCommand.temperatureChange))) 
    .thenReply(_ => Empty.defaultInstance) 
}

And finally, let's update the event handler to account for the different types of temperature changes.

override def temperatureChange(currentState: TurkeyState, temperatureChange: TemperatureChange): TurkeyState = {
    temperatureChange.changeType match {
    case TemperatureChange.Type.EXTERNAL => currentState.copy(externalTemperature = temperatureChange.newTemperature)
    case TemperatureChange.Type.INTERNAL => currentState.copy(internalTemperature = temperatureChange.newTemperature)
    }    
}

Smell that? That's some turkey cooking! Let's see how far along it is.

Re-start the local dev environment and make some API calls

In that same terminal command window, you can CTRL+C to stop the current running service. If not already running, simply type start.sh, which will recompile all of the protobufs, stop any running associated Docker containers and start the updated Akka Serverless service.

And now for some cooking! I mean...curling!

curl -X POST -H 'Content-Type: application/json' http://localhost:9000/com.example.TurkeyService/StartCooking -d '{"turkey_id": "myfirstturkey"}'

curl -X POST -H 'Content-Type: application/json' http://localhost:9000/com.example.TurkeyService/IncreaseOvenTemperature -d '{"turkey_id": "myfirstturkey", "temperature_change": 100.0}'

curl -X POST -H 'Content-Type: application/json' http://localhost:9000/com.example.TurkeyService/GetCurrentTurkey -d '{"turkey_id": "myfirstturkey"}'

The response should be:

{"inOven":true,"done":"RAW","externalTemperature":100.0}%

Let's make some additional API calls.

    curl -i -v -X POST -H 'Content-Type: application/json' http://localhost:9000/com.example.TurkeyService/IncreaseTurkeyTemperature -d '{"turkey_id": "myfirstturkey", "temperature_change": 10.0}'

    curl -X POST -H 'Content-Type: application/json' http://localhost:9000/com.example.TurkeyService/GetCurrentTurkey -d '{"turkey_id": "myfirstturkey"}'

The response should be:

{"inOven":true,"done":"RAW","externalTemperature":100.0,"internalTemperature":10.0}%

Before moving on to the next section of this post, please shut down the services. CTRL+C should work but it would be good to ensure proper shutdown of the Docker containers spun up: docker-compose -f docker-compose-proxy.yml down.

Integrating Kafka events into our service

NOTE: You can git clone kafka_integration to get all of the code changes below.

One of the amazing features in Akka Serverless is the ability to ingest (and egress if so desired) events from external messaging systems, e.g. Google Pubsub and Kafka, as well as use events to communicate between services running in the Akka Serverless Platform-as-a-Service (PaaS). You can read more about this feature here.

Eventing easy to add though! And actually won't require any more Scala coding beyond the already implemented change for tracking a turkey's internal temperature.

First, let's define that events will be supported, and of what type, for our APIs. For our prevent-salmonella-poisoning scenario, let's imagine that industry - a company perhaps - has delivered to the market a product that enables for remote recommendation generation based on machine-learning models being run in a cloud distant from the homes in which the turkeys are being cooked. We're not going to actually create an ML model and run it but we will simulate the process by which those events could be published on to a company's Kafka broker and sent through the pipes into Akka Serverless. So instead of making gRPC or HTTP API calls to increase the temperature of the oven, for example, we can have those commands issued from deep within the bowels of a company's infrastructure and sent over Kafka to the turkey entities managed in the Akka Serverless cloud. Neat, huh?

To do that, we will update the commands in the turkey_api.proto: IncreaseOvenTemperature and DecreaseOvenTemperature. We can add the protobuf annotations supplied by Akka Serverless, to say "when TemperatureChangeCommands are received on the increase_temp topic, execute the IncreaseOvenTemperature command." We do that for both commands.

rpc IncreaseOvenTemperature(TemperatureChangeCommand) returns (google.protobuf.Empty) {
    option (akkaserverless.method).eventing.in = {
        topic: "increase_temp"
    };
}
rpc DecreaseOvenTemperature(TemperatureChangeCommand) returns (google.protobuf.Empty) {
    option (akkaserverless.method).eventing.in = {
        topic: "decrease_temp"
    };  
}

That is it. I did say it was neat, didn't I?

To run this locally, we take advantage of a great product, Redpanda from Vectorized.io. They have a Kafka API compatible streaming platform for mission-critical workloads, and a great way to run that locally (Docker or not).

In a terminal window, in the root of the project directory: docker-compose -f docker-compose-proxy.yml. This will start both the Akka Serverless proxy as well as the Redpanda container. From another terminal window, run the sbt run command. You will notice some error messages that indicate that the topics, increase_temp and decrease_temp need to be created. Ignore for now, given that we are running locally. When moving to a hosted offering, like Confluent Cloud you will need to create topics manually.

Make some API calls and issue Kafka events

First, let's get the initial state of our turkey.

curl -X GET -H 'Content-Type: application/json' http://localhost:9000/turkeys/myfirstturkey

The result should look like.

{"inOven":false,"done":"RAW","internalTemperature":0.0,"externalTemperature":0.0}%

From a new terminal window, in the root directory of the project: sbt "runMain com.example.KafkaMessageGenerator myfirstturkey increase 10". This will fire a TemperatureChangeCommand event on the increase_temp topic, sent to the locally running Redpanda cluster.

Now if we make that same GET call again, from above.

curl -X GET -H 'Content-Type: application/json' http://localhost:9000/turkeys/myfirstturkey

The result should look like.

{"inOven":false,"done":"RAW","internalTemperature":10.0,"externalTemperature":0.0}%

We have successfully delivered event data from an external source, via Kafka (Redpanda in this local case), into Akka Serverless. Profit!

Where to next?

Install the Akka Serverless CLI;
Once installed, you can sign-up for a new account via akkasls auth signup (or visit the Console Sign-up page;
Login to your account via the CLI: akkasls auth login;
In a terminal window, in the root directory of the project, package up the docker container and publish: sbt docker:publish -Ddocker.username=<insert docker registry username>.
Configure your Kakfa broker per the [documentation](https://developer.lightbend.com/docs/akka-serverless/projects/message-brokers.html#_confluent_cloud. Do not pick the Python client, since the actual Kafka integration is running in the Akka Serverless proxy.
Per the same documentation, create the increase_temp and decrease_temp topics in Confluent Cloud.
In the same terminal window, run the command akkasls services deploy turkey-tracker $DOCKER_REGISTRY/$DOCKER_USER/turkey_tracker_scala:<insert tag from step #4 above
In the same terminal window, run the command akkasls services proxy turkey-tracker --grpcui. You can use this to explore the API running in the cloud without exposing over the internet.

Conclusion

Hopefully you now have a better idea on how some of the moving parts work in Akka Serverless and how you can build event driven APIs and services, including those that can be integrated with Kafka. While we're not really cooking turkeys, perhaps one day, we can leverage technologies like this to improve all of our Thanksgiving food-eating experiences!

And don't forget: get your free Akka Serverless account here.

Photo credit: https://www.istockphoto.com/portfolio/AlexRaths?mediatype=photography