Lagomのメッセージブローカーの完全な例

Question

私はLagom 1.2.2でセットアップされたメッセージブローカーを実装しようとしていて、壁に入っています。ドキュメントには、サービス記述子のために、次の例があります。

実装の

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オブジェクトを消費していているようですが、私はGreetingMessageオブジェクトを返すように機能convertEventを作成するときに、私はコンパイルエラーを取得：

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サンプルアプリをチェックしてあり、これの例はないようです。

ありがとうございます！あなたが貼り付けられた

Answer 1

エラーメッセージが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.java#L91