Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Make message optional in topic producer event stream #2708

Open
wants to merge 15 commits into
base: master
Choose a base branch
from
Open

Make message optional in topic producer event stream #2708

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
15 commits
Select commit Hold shift + click to select a range
47aff7e
make message optional in topic producer event stream
bossqone Feb 17, 2020
280fb12
Revert "make message optional in topic producer event stream"
bossqone May 7, 2020
aa4a2d5
Merge branch 'master' into topic-producer-optional-message
bossqone May 7, 2020
fd46db2
introduce topic producer API with commands EmitAndCommit, Commit [WIP]
bossqone May 15, 2020
baaed46
fix formatting
bossqone May 15, 2020
361b9ee
fix compilation
bossqone May 15, 2020
19391ef
fixed implementation of TopicProducerActor
bossqone May 19, 2020
e78966d
Merge branch 'master' into topic-producer-optional-message
bossqone Jul 17, 2020
f35a811
renamed methods
bossqone Jul 20, 2020
7e7dcec
added support for EmitMultipleAndCommit
bossqone Jul 21, 2020
415e978
renamed methods
bossqone Jul 21, 2020
0e63460
Revert "added support for EmitMultipleAndCommit"
bossqone Aug 28, 2020
79f3575
Merge branch 'master' into topic-producer-optional-message
bossqone Aug 28, 2020
91ca90c
changed TopicProducerCommand case class to class + fixed tests
bossqone Aug 29, 2020
ee5b7ae
updated .gitignore
bossqone Aug 29, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions ...rc/test/scala/com/lightbend/lagom/internal/javadsl/broker/kafka/JavadslKafkaApiSpec.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -247,8 +247,8 @@ class JavadslKafkaApiSpec

// Put some messages in the stream
test3EventJournal.append(("firstMessage", true))
test3EventJournal.append(("secondMessage", true))
test3EventJournal.append(("thirdMessage", false))
test3EventJournal.append(("secondMessage", false))
test3EventJournal.append(("thirdMessage", true))

// Wait for a subscriber to consume them all (which ensures they've all been published)
val allMessagesReceived = new CountDownLatch(2)
ignasi35 marked this conversation as resolved.
Show resolved Hide resolved
Expand Down
2 changes: 1 addition & 1 deletion service/scaladsl/broker/src/main/scala/com/lightbend/internal/broker/TopicProducers.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,7 +41,7 @@ object TaggedOffsetTopicProducer {
new TaggedOffsetTopicProducer[Message, Event](
tags,
(tag, offset) =>
readSideStream.apply(tag, offset).map(pair => TopicProducerCommand.EmitAndCommit(pair._1, pair._2))
readSideStream.apply(tag, offset).map(pair => new TopicProducerCommand.EmitAndCommit(pair._1, pair._2))
)

}
4 changes: 3 additions & 1 deletion ...ce/scaladsl/broker/src/main/scala/com/lightbend/lagom/scaladsl/broker/TopicProducer.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -81,7 +81,9 @@ object TopicProducer {

// TODO(bossqone): add docs
@ApiMayChange
def taggedCommandStreamWithOffset[Message, Event <: AggregateEvent[Event]](tags: immutable.Seq[AggregateEventTag[Event]])(
def taggedCommandStreamWithOffset[Message, Event <: AggregateEvent[Event]](
tags: immutable.Seq[AggregateEventTag[Event]]
)(
eventStream: (AggregateEventTag[Event], Offset) => Source[TopicProducerCommand[Message], Any]
): Topic[Message] = TaggedOffsetTopicProducer.fromTopicProducerCommandStream(tags, eventStream)

Expand Down
20 changes: 18 additions & 2 deletions ...adsl/broker/src/main/scala/com/lightbend/lagom/scaladsl/broker/TopicProducerCommand.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,11 +4,27 @@

package com.lightbend.lagom.scaladsl.broker

import java.util.Objects;
import akka.persistence.query.Offset

sealed trait TopicProducerCommand[Message]

object TopicProducerCommand {
case class EmitAndCommit[Message](message: Message, offset: Offset) extends TopicProducerCommand[Message]
case class Commit[Message](offset: Offset) extends TopicProducerCommand[Message]
final class EmitAndCommit[Message](val message: Message, val offset: Offset) extends TopicProducerCommand[Message] {
override def equals(that: Any): Boolean = that match {
case command: EmitAndCommit[Message] => message.equals(command.message) && offset.equals(command.offset)
case _ => false
}

override def hashCode(): Int = Objects.hash(message.asInstanceOf[AnyRef], offset)
}

final class Commit[Message](val offset: Offset) extends TopicProducerCommand[Message] {
override def equals(that: Any): Boolean = that match {
case command: Commit[Message] => offset.equals(command.offset)
case _ => false
}

override def hashCode(): Int = Objects.hash(offset)
}
}
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Looks great.

As this is public API there should be API docs.

In future improvements, doesn't have to be in this PR, we could add a toString and also some factory methods in object TopicProducerCommand so building instances of EmitAndCommit is shorter than new TopicProducerCommand.EmitAndCommit.

6 changes: 3 additions & 3 deletions ...a/com/lightbend/lagom/internal/scaladsl/broker/kafka/ScaladslRegisterTopicProducers.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -118,8 +118,8 @@ class ScaladslRegisterTopicProducers(
command: TopicProducerCommand[Message]
): InternalTopicProducerCommand[Message] =
command match {
case TopicProducerCommand.EmitAndCommit(message, offset) =>
InternalTopicProducerCommand.EmitAndCommit(message, offset)
case TopicProducerCommand.Commit(offset) => InternalTopicProducerCommand.Commit(offset)
case cmd: TopicProducerCommand.EmitAndCommit[Message] =>
InternalTopicProducerCommand.EmitAndCommit(cmd.message, cmd.offset)
case cmd: TopicProducerCommand.Commit[Message] => InternalTopicProducerCommand.Commit(cmd.offset)
}
}
8 changes: 4 additions & 4 deletions ...erver/src/test/scala/com/lightbend/lagom/scaladsl/kafka/broker/ScaladslKafkaApiSpec.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -235,8 +235,8 @@ class ScaladslKafkaApiSpec

// Put some messages in the stream
test3EventJournal.append(("firstMessage", true))
test3EventJournal.append(("secondMessage", true))
test3EventJournal.append(("thirdMessage", false))
test3EventJournal.append(("secondMessage", false))
test3EventJournal.append(("thirdMessage", true))

// Wait for a subscriber to consume them all (which ensures they've all been published)
val allMessagesReceived = new CountDownLatch(2)
ignasi35 marked this conversation as resolved.
Show resolved Hide resolved
Expand Down Expand Up @@ -421,8 +421,8 @@ object ScaladslKafkaApiSpec {
.eventStream(fromOffset)
.map {
case ((message, emit), offset) if emit =>
TopicProducerCommand.EmitAndCommit(messageTransformer(message), offset)
case (_, offset) => TopicProducerCommand.Commit(offset)
new TopicProducerCommand.EmitAndCommit(messageTransformer(message), offset)
case (_, offset) => new TopicProducerCommand.Commit(offset)
}
}
}
Expand Down
2 changes: 1 addition & 1 deletion ...it/scaladsl/src/main/scala/com/lightbend/lagom/scaladsl/testkit/TestTopicComponents.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -96,7 +96,7 @@ private[lagom] class TestTopic[Payload, Event <: AggregateEvent[Event]](
topicProducer.readSideStream.apply(tag, Offset.noOffset)
})
.collect {
case TopicProducerCommand.EmitAndCommit(message, _) => message
case cmd: TopicProducerCommand.EmitAndCommit[Payload] => cmd.message
}
.map { evt =>
serializer.serializerForRequest.serialize(evt)
Expand Down