0

我正在尝试实现一个使用 Lagom 1.2.2 设置的 Message Broker，但遇到了麻烦。该文档具有以下服务描述符示例：

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

这个实现示例：

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

但是，没有示例说明函数的参数类型或返回类型是什么convertEvent()，这就是我要画一个空白的地方。另一方面，MessageBroker 的订阅者，它似乎正在使用GreetingMessage对象，但是当我创建一个convertEvent返回GreetingMessage对象的函数时，我得到一个编译错误：

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)

有没有更彻底的例子来说明如何使用它？我已经检查了 Chirper 示例应用程序，它似乎没有这样的示例。

谢谢！

4

1 回答 1

3

您粘贴的错误消息准确地告诉您map预期的内容：

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

所以，你需要传递一个接受Pair<GreetingEvent, Offset>. 函数应该返回什么？好吧，更新它来接受它，然后你会得到下一个错误，它会再次告诉你它期望你返回什么，在这种情况下你会发现它是Pair<GreetingMessage, Offset>.

为了解释这些类型是什么 - Lagom 需要跟踪哪些事件已发布到 Kafka，这样当您重新启动服务时，它不会从事件日志的开头开始并重新从时间的开头重新发布所有事件. 它通过使用偏移量来做到这一点。所以事件日志会产生成对的事件和偏移量，然后你需要将这些事件转换成将要发布到Kafka的消息，并且当你将转换后的消息返回给Lagom时，它需要与偏移量成对您从事件日志中获得的，以便在发布到 Kafka 后，Lagom 可以保留偏移量，并在下次重新启动服务时将其用作起点。

一个完整的例子可以在这里看到：https://github.com/lagom/online-auction-java/blob/a32e696/bidding-impl/src/main/java/com/example/auction/bidding/impl/BiddingServiceImpl。爪哇#L91

于 2017-02-21T05:37:09.433 回答