Full example of Message Broker in Lagom

I'm trying to implement a Message Broker set up with Lagom 1.2.2 and have run into a wall. The documentation has the following example for the service descriptor:

default Descriptor descriptor() {
return named("helloservice").withCalls(...)
  // here we declare the topic(s) this service will publish to
  .publishing(
    topic("greetings", this::greetingsTopic)
  )
  ....;
}

And this example for the implementation:

public Topic<GreetingMessage> greetingsTopic() {
return TopicProducer.singleStreamWithOffset(offset -> {
    return persistentEntityRegistry
        .eventStream(HelloEventTag.INSTANCE, offset)
        .map(this::convertEvent);
  });
}

However, there's no example of what the argument type or return type of the convertEvent() function are, and this is where I'm drawing a blank. On the other end, the subscriber to the MessageBroker, it seems that it's consuming GreetingMessage objects, but when I create a function convertEvent to return GreetingMessage objects, I get a compilation error:

Error:(61, 21) java: method map in class akka.stream.javadsl.Source<Out,Mat> cannot be applied to given types;
  required: akka.japi.function.Function<akka.japi.Pair<com.example.GreetingEvent,com.lightbend.lagom.javadsl.persistence.Offset>,T>
  found: this::convertEvent
  reason: cannot infer type-variable(s) T
    (argument mismatch; invalid method reference
  incompatible types: akka.japi.Pair<com.example.GreetingEvent,com.lightbend.lagom.javadsl.persistence.Offset> cannot be converted to com.example.GreetingMessage)

Are there any more more thorough examples of how to use this? I've already checked in the Chirper sample app and it doesn't seem to have an example of this.

Thanks!

Solution

The error message you pasted tells you exactly what map expects:

required: akka.japi.function.Function<akka.japi.Pair<com.example.GreetingEvent,com.lightbend.lagom.javadsl.persistence.Offset>,T>

So, you need to pass a function that takes Pair<GreetingEvent, Offset>. What should the function return? Well, update it to take that, and then you'll get the next error, which once again will tell you what it was expecting you to return, and in this instance you'll find it's Pair<GreetingMessage, Offset>.

To explain what these types are - Lagom needs to track which events have been published to Kafka, so that when you restart a service, it doesn't start from the beginning of your event log and republish all the events from the beginning of time again. It does this by using offsets. So the event log produces pairs of events and offsets, and then you need to transform these events to the messages that will be published to Kafka, and when you returned the transformed message to Lagom, it needs to be a in a pair with the offset that you got from the event log, so that after publishing to Kafka, Lagom can persist the offset, and use that as the starting point next time the service is restarted.

A full example can be seen here: https://github.com/lagom/online-auction-java/blob/a32e696/bidding-impl/src/main/java/com/example/auction/bidding/impl/BiddingServiceImpl.java#L91