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

Elasticsearch: use proper FlowWithContext #1766

Merged
merged 2 commits into from
Jun 20, 2019
Merged

Elasticsearch: use proper FlowWithContext #1766

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
1aeb277
Elasticsearch: use proper FlowWithContext
ennru Jun 20, 2019
b2c1d5b
Add MiMa exclusion
ennru Jun 20, 2019
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
3 changes: 3 additions & 0 deletions elasticsearch/src/main/mima-filters/1.0.2.backwards.excludes
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,3 @@
# PR #1766 use proper FlowWithContext
ProblemFilters.exclude[IncompatibleResultTypeProblem]("akka.stream.alpakka.elasticsearch.javadsl.ElasticsearchFlow.createWithContext")
ProblemFilters.exclude[IncompatibleResultTypeProblem]("akka.stream.alpakka.elasticsearch.scaladsl.ElasticsearchFlow.createWithContext")
14 changes: 5 additions & 9 deletions ...icsearch/src/main/scala/akka/stream/alpakka/elasticsearch/javadsl/ElasticsearchFlow.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,6 @@ package akka.stream.alpakka.elasticsearch.javadsl

import akka.NotUsed
import akka.annotation.ApiMayChange
import akka.japi.Pair
import akka.stream.alpakka.elasticsearch.{impl, _}
import akka.stream.scaladsl
import com.fasterxml.jackson.databind.ObjectMapper
Expand Down Expand Up @@ -89,12 +88,9 @@ object ElasticsearchFlow {
settings: ElasticsearchWriteSettings,
elasticsearchClient: RestClient,
objectMapper: ObjectMapper
): akka.stream.javadsl.Flow[Pair[WriteMessage[T, NotUsed], C], Pair[WriteResult[T, C], C], NotUsed] =
): akka.stream.javadsl.FlowWithContext[WriteMessage[T, NotUsed], C, WriteResult[T, C], C, NotUsed] =
scaladsl
.Flow[Pair[WriteMessage[T, NotUsed], C]]
.map { pair =>
pair.first.withPassThrough(pair.second)
}
.Flow[WriteMessage[T, C]]
.batch(settings.bufferSize, immutable.Seq(_)) { case (seq, wm) => seq :+ wm }
.via(
new impl.ElasticsearchFlowStage[T, C](indexName,
Expand All @@ -104,9 +100,9 @@ object ElasticsearchFlow {
new JacksonWriter[T](objectMapper))
)
.mapConcat(identity)
.map { wr =>
Pair.create(wr, wr.message.passThrough)
}
.asFlowWithContext[WriteMessage[T, NotUsed], C, C]((res, c) => res.withPassThrough(c))(
p => p.message.passThrough
)
.asJava

private final class JacksonWriter[T](mapper: ObjectMapper) extends MessageWriter[T] {
Expand Down
18 changes: 7 additions & 11 deletions ...csearch/src/main/scala/akka/stream/alpakka/elasticsearch/scaladsl/ElasticsearchFlow.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,7 +8,7 @@ import akka.NotUsed
import akka.annotation.ApiMayChange
import akka.stream.alpakka.elasticsearch._
import akka.stream.alpakka.elasticsearch.impl
import akka.stream.scaladsl.Flow
import akka.stream.scaladsl.{Flow, FlowWithContext}
import org.elasticsearch.client.RestClient
import spray.json._

Expand Down Expand Up @@ -120,7 +120,7 @@ object ElasticsearchFlow {
settings: ElasticsearchWriteSettings = ElasticsearchWriteSettings())(
implicit elasticsearchClient: RestClient,
sprayJsonWriter: JsonWriter[T]
): Flow[(WriteMessage[T, NotUsed], C), (WriteResult[T, C], C), NotUsed] =
): FlowWithContext[WriteMessage[T, NotUsed], C, WriteResult[T, C], C, NotUsed] =
createWithContext[T, C](indexName, typeName, settings, new SprayJsonWriter[T]()(sprayJsonWriter))

/**
Expand All @@ -137,18 +137,14 @@ object ElasticsearchFlow {
settings: ElasticsearchWriteSettings,
writer: MessageWriter[T])(
implicit elasticsearchClient: RestClient
): Flow[(WriteMessage[T, NotUsed], C), (WriteResult[T, C], C), NotUsed] = {
): FlowWithContext[WriteMessage[T, NotUsed], C, WriteResult[T, C], C, NotUsed] = {
require(settings.retryLogic == RetryNever,
"`withContext` may not be used with retrying enabled, as it disturbs element order")
Flow[(WriteMessage[T, NotUsed], C)]
.map {
case (wm, pt) =>
wm.withPassThrough(pt)
}
Flow[WriteMessage[T, C]]
.via(createWithPassThrough(indexName, typeName, settings, writer))
.map { wr =>
(wr, wr.message.passThrough)
}
.asFlowWithContext[WriteMessage[T, NotUsed], C, C]((res, c) => res.withPassThrough(c))(
p => p.message.passThrough
)
}

private final class SprayJsonWriter[T](implicit writer: JsonWriter[T]) extends MessageWriter[T] {
Expand Down