Implementing Actions in Java

Language

Actions are stateless functions that can be used to implement different uses cases, such as:

a pure function
request conversion - you can use Actions to convert incoming data into a different format before forwarding a call to a different component.
publish and subscribe to events
schedule and cancel timers

Actions can be triggered in multiple ways. For example, by:

a gRPC service call
an HTTP service call
a forwarded call from another component
a scheduled call from a timer
an incoming event from within the same service or a from different service

Defining the `proto` file

An Action may implement any service method defined in a Protobuf definition. In this first example, we will show how to implement an Action as a pure stateless function. We will define a FibonacciAction that takes a number and return the next number in the Fibonacci series.

Java

src/main/proto/com/example/fibonacci/fibonacci.proto

syntax = "proto3";
package com.example.fibonacci; (1)

import "kalix/annotations.proto"; (2)

option java_outer_classname = "FibonacciApi"; (3)

message Number {
  int64 value = 1;
}

service Fibonacci {
  option (kalix.codegen) = {
    action: {}  (4)
  };

  rpc NextNumber(Number) returns (Number) {}
}

1	Any classes generated from this protobuf file will be in the `com.example.fibonacci` package.
2	Import the Kalix protobuf annotations or options.
3	Let the messages declared in this protobuf file be inner classes to the Java class `FibonacciApi`.
4	The protobuf option (kalix.codegen) is specific to code-generation as provided by the Kalix Maven plugin. This annotation indicates to the code-generation that an Action must be generated.

Scala

src/main/proto/com/example/fibonacci/fibonacci.proto

syntax = "proto3";
package com.example.fibonacci; (1)

import "kalix/annotations.proto"; (2)

message Number {
  int64 value = 1;
}

service Fibonacci {
  option (kalix.codegen) = {
    action: {}  (3)
  };

  rpc NextNumber(Number) returns (Number) {}
}

1	Any classes generated from this protobuf file will be in the `com.example.fibonacci` package.
2	Import the Kalix protobuf annotations or options.
3	The protobuf option (kalix.codegen) is specific to code-generation as provided by the Kalix sbt plugin. This annotation indicates to the code-generation that an Action must be generated.

Implementing the Action

An Action implementation is a class where you define how each message is handled. The class FibonacciAction gets generated for us based on the proto file defined above. Once the FibonacciAction.java FibonacciAction.scala file exists, it is not overwritten, so you can freely add logic to it. FibonacciAction extends the generated class AbstractFibonacciAction which we’re not supposed to change as it gets regenerated in case we update the protobuf descriptors.

AbstractFibonacciAction contains all method signatures corresponding to the API of the service. If you change the API you will see compilation errors in the FibonacciAction class, and you have to implement the methods required by AbstractFibonacciAction.

Java

src/main/java/com/example/fibonacci/FibonacciAction.java

public class FibonacciAction extends AbstractFibonacciAction { (1)

  public FibonacciAction(ActionCreationContext creationContext) {
  }

  /**
   * Handler for "NextNumber".
   */
  @Override
  public Effect<FibonacciApi.Number> nextNumber(FibonacciApi.Number number) { (2)
    throw new RuntimeException("The command handler for `NextNumber` is not implemented, yet");
  }

}

1	Extends the generated `AbstractFibonacciAction`, which extends `Action` .
2	A `nextNumber` method is generated. We will implement it next.

Scala

src/main/scala/com/example/fibonacci/FibonacciAction.scala

class FibonacciAction(creationContext: ActionCreationContext) extends AbstractFibonacciAction {  (1)

  override def nextNumber(number: Number): Action.Effect[Number] = { (2)
    throw new RuntimeException("The command handler for `NextNumber` is not implemented, yet")
  }
}

1	Extends the generated `AbstractFibonacciAction`, which extends `Action` .
2	A `nextNumber` method is generated. We will implement it next.

Next, we can implement nextNumber method to complete our Action.

Java

src/main/java/com/example/fibonacci/FibonacciAction.java

private boolean isFibonacci(long num) {  (1)
  Predicate<Long> isPerfectSquare = (n) -> {
    long square = (long) Math.sqrt(n);
    return square*square == n;
  };
  return isPerfectSquare.test(5*num*num + 4) || isPerfectSquare.test(5*num*num - 4);
}
private long nextFib(long num) {
  double result = num * (1 + Math.sqrt(5)) / 2.0;
  return Math.round(result);
}

@Override
public Effect<FibonacciApi.Number> nextNumber(FibonacciApi.Number number) {
  long num = number.getValue();
  if (isFibonacci(num)) { (2)
    long nextFib = nextFib(num);
    FibonacciApi.Number response =
        FibonacciApi.Number
            .newBuilder()
            .setValue(nextFib)
            .build();
    return effects().reply(response);
  } else {
    return effects() (3)
             .error("Input number is not a Fibonacci number, received '" + num + "'");
  }
}

1	We add two private methods to support the computation. `isFibonacci` checks if a number is a Fibonacci number and `nextFib` calculates the next number.
2	The `nextNumber` implementation first checks if the input number belongs to the Fibonacci series. If so, it calculates the next number and builds a reply using `effects().reply()`.
3	Otherwise, if the input number doesn’t belong to the Fibonacci series, it builds an `Effect` reply error.

Scala

src/main/scala/com/example/fibonacci/FibonacciAction.scala

private def isFibonacci(num: Long): Boolean = { (1)
  val isPerfectSquare = (n: Long) => {
    val square = Math.sqrt(n.toDouble).toLong
    square * square == n
  }
  isPerfectSquare(5 * num * num + 4) || isPerfectSquare(5 * num * num - 4)
}

private def nextFib(num: Long): Long = {
  val result = num * (1 + Math.sqrt(5)) / 2.0;
  Math.round(result)
}

override def nextNumber(number: Number): Action.Effect[Number] = {
  val num = number.value
  if (isFibonacci(num)) (2)
    effects.reply(Number(nextFib(num)))
  else
    effects.error(s"Input number is not a Fibonacci number, received '$num'") (3)
}

1	We add two private methods to support the computation. `isFibonacci` checks if a number is a Fibonacci number and `nextFib` calculates the next number.
2	The `nextNumber` implementation first checks if the input number belongs to the Fibonacci series. If so, it calculates the next number and builds a reply using `effects.reply()`.
3	Otherwise, if the input number doesn’t belong to the Fibonacci series, it builds an `Effect` reply error.

Multiple replies / reply streaming

An Action may return data conditionally by marking the return type as stream in Protobuf. The Java method implementing that service must return an Akka Streams Source to fulfill that contract.

The Source may publish an arbitrary number of replies.

Registering the Action

To make Kalix aware of the Action, we need to register it with the service.

From the code-generation, the registration gets automatically inserted in the generated KalixFactory.withComponents method from the Main class.

Java

/src/main/java/com/example/Main.java

/* This code was generated by Kalix tooling.
 * As long as this file exists it will not be re-generated.
 * You are free to make changes to this file.
 */

package com.example;

import kalix.javasdk.Kalix;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.example.fibonacci.FibonacciAction;

public final class Main {

  private static final Logger LOG = LoggerFactory.getLogger(Main.class);

  public static Kalix createKalix() {
    // The KalixFactory automatically registers any generated Actions, Views or Entities,
    // and is kept up-to-date with any changes in your protobuf definitions.
    // If you prefer, you may remove this and manually register these components in a
    // `new Kalix()` instance.
    return KalixFactory.withComponents(
      FibonacciAction::new);
  }

  public static void main(String[] args) throws Exception {
    LOG.info("starting the Kalix service");
    createKalix().start();
  }
}

Scala

/src/main/scala/com/example/fibonacci/Main.scala

package com.example.fibonacci

import kalix.scalasdk.Kalix
import org.slf4j.LoggerFactory

object Main {

  private val log = LoggerFactory.getLogger("com.example.fibonacci.Main")

  def createKalix(): Kalix = {
    // The KalixFactory automatically registers any generated Actions, Views or Entities,
    // and is kept up-to-date with any changes in your protobuf definitions.
    // If you prefer, you may remove this and manually register these components in a
    // `Kalix()` instance.
    KalixFactory.withComponents(
      new FibonacciAction(_))
  }

  def main(args: Array[String]): Unit = {
    log.info("starting the Kalix service")
    createKalix().start()
  }
}

By default, the generated constructor has an ActionCreationContext parameter, but you can change this to accept other parameters. If you change the constructor of the FibonacciAction class you will see a compilation error here, and you have to adjust the factory function that is passed to KalixFactory.withComponents.

When more components are added the KalixFactory is regenerated, and you have to adjust the registration from the Main class.

Testing the Action

Unit tests

The following snippet shows how the FibonacciActionTestKit is used to test the FibonacciAction implementation.

Kalix generates the FibonacciActionTestKit that allows us to call the methods of FibonacciAction. For each Action Kalix generates a specific test kit for it, with the name [ActionName]TestKit. Each call we execute over to the test kit returns an ActionResult that holds the effect produced by the underlying action method.

Apart from the test kit Kalix generates test classes based on the Action defined in the .proto files. This is shown in the snippet below.

Actions are unique units of computation where no local state is shared with previous or subsequent calls. The framework doesn’t reuse an Action instance but creates a new one for each command handled and therefore it is also how the test kit behaves.

Java

src/test/java/com/example/actions/FibonacciActionTest.java

public class FibonacciActionTest {

  @Test
  public void nextNumberTest() {
    FibonacciActionTestKit testKit = FibonacciActionTestKit.of(FibonacciAction::new); (1)
    ActionResult<FibonacciApi.Number> result = testKit.nextNumber(FibonacciApi.Number.newBuilder().setValue(5).build()); (2)
    assertEquals(8, result.getReply().getValue()); (3)
  }

}

1	The test kit is created to allow us to test the Action’s method.
2	We call `nextNumber` method with some value.
3	The reply message from the result is retrieved using `getReply()`.

ActionResult

Calling an action method through the test kit gives us back an ActionResult . This class has methods that we can use to assert our expectations, such as:

getReply() returns the reply message passed to effects().reply() or throws an exception failing the test, if the effect returned was not a reply.
getError() returns the error description when effects().error() was returned to signal an error.
getForward() returns details about what message was forwarded and where the call was forwarded (since it is a unit test the forward is not actually executed).

Scala

src/test/java/com/example/actions/FibonacciActionSpec.scala

class FibonacciActionSpec
  extends AnyWordSpec
    with ScalaFutures
    with Matchers {

  "FibonacciAction" must {

    "handle command NextNumber" in {
      val testKit = FibonacciActionTestKit(new FibonacciAction(_)) (1)
      val result = testKit.nextNumber(Number(5)) (2)
      result.reply shouldBe (Number(8)) (3)
    }
  }
}

1	The test kit is created to allow us to test the Action’s method.
2	We call `nextNumber` method with some value.
3	The reply message from the result is retrieved using `reply`.

ActionResult

Calling an action method through the test kit gives us back an ActionResult . This class has methods that we can use to assert our expectations, such as:

reply returns the reply message passed to effects.reply() or throws an exception failing the test, if the effect returned was not a reply.
errorDescription returns the error description when effects().error() was returned to signal an error.
forwardedTo returns details about what message was forwarded and where the call was forwarded (since it is a unit test the forward is not actually executed).

By default the integration and unit test are both invoked by sbt test. To only run unit tests run sbt -DonlyUnitTest test, or sbt -DonlyUnitTest=true test, or set up that value to true in the sbt session by set onlyUnitTest := true and then run test

Implementing Actions in Java

Defining the proto file

Implementing the Action

Multiple replies / reply streaming

Registering the Action

Testing the Action

Unit tests

Defining the `proto` file