diff --git a/runtime/src/main/mima-filters/2.4.0-M1.backwards.excludes/1838-deadline-per-call.excludes b/runtime/src/main/mima-filters/2.4.0-M1.backwards.excludes/1838-deadline-per-call.excludes new file mode 100644 index 000000000..445d65b8a --- /dev/null +++ b/runtime/src/main/mima-filters/2.4.0-M1.backwards.excludes/1838-deadline-per-call.excludes @@ -0,0 +1,6 @@ +# New APIs for setting deadline per call + +ProblemFilters.exclude[ReversedMissingMethodProblem]("akka.grpc.javadsl.SingleResponseRequestBuilder.setDeadline") +ProblemFilters.exclude[ReversedMissingMethodProblem]("akka.grpc.javadsl.StreamResponseRequestBuilder.setDeadline") +ProblemFilters.exclude[ReversedMissingMethodProblem]("akka.grpc.scaladsl.SingleResponseRequestBuilder.setDeadline") +ProblemFilters.exclude[ReversedMissingMethodProblem]("akka.grpc.scaladsl.StreamResponseRequestBuilder.setDeadline") diff --git a/runtime/src/main/scala/akka/grpc/internal/RequestBuilderImpl.scala b/runtime/src/main/scala/akka/grpc/internal/RequestBuilderImpl.scala index 02418fea5..e1fcf7b9a 100644 --- a/runtime/src/main/scala/akka/grpc/internal/RequestBuilderImpl.scala +++ b/runtime/src/main/scala/akka/grpc/internal/RequestBuilderImpl.scala @@ -4,11 +4,12 @@ package akka.grpc.internal -import java.util.concurrent.CompletionStage - +import java.time.{ Duration => JDuration } +import java.util.concurrent.{ CompletionStage, TimeUnit } import akka.NotUsed import akka.annotation.{ InternalApi, InternalStableApi } import akka.dispatch.ExecutionContexts +import akka.grpc.scaladsl.SingleResponseRequestBuilder import akka.grpc.{ GrpcClientSettings, GrpcResponseMetadata, GrpcServiceException, GrpcSingleResponse } import akka.stream.{ Graph, Materializer, SourceShape } import akka.stream.javadsl.{ Source => JavaSource } @@ -17,6 +18,7 @@ import akka.util.ByteString import io.grpc._ import scala.compat.java8.FutureConverters._ +import scala.concurrent.duration.Duration import scala.concurrent.{ ExecutionContext, Future } /** @@ -52,6 +54,15 @@ final class ScalaUnaryRequestBuilder[I, O]( override def withHeaders(headers: MetadataImpl): ScalaUnaryRequestBuilder[I, O] = new ScalaUnaryRequestBuilder[I, O](descriptor, channel, defaultOptions, settings, headers) + + override def setDeadline(deadline: Duration): SingleResponseRequestBuilder[I, O] = + new ScalaUnaryRequestBuilder[I, O]( + descriptor, + channel, + if (!deadline.isFinite) defaultOptions.withDeadline(null) + else defaultOptions.withDeadlineAfter(deadline.toMillis, TimeUnit.MILLISECONDS), + settings, + headers) } /** @@ -84,6 +95,15 @@ final class JavaUnaryRequestBuilder[I, O]( override def withHeaders(headers: MetadataImpl): JavaUnaryRequestBuilder[I, O] = new JavaUnaryRequestBuilder[I, O](descriptor, channel, defaultOptions, settings, headers) + + override def setDeadline(deadline: JDuration): JavaUnaryRequestBuilder[I, O] = + new JavaUnaryRequestBuilder[I, O]( + descriptor, + channel, + if (deadline == null) defaultOptions.withDeadline(null) + else defaultOptions.withDeadlineAfter(deadline.toMillis, TimeUnit.MILLISECONDS), + settings, + headers) } /** @@ -152,6 +172,15 @@ final class ScalaClientStreamingRequestBuilder[I, O]( override def withHeaders(headers: MetadataImpl): ScalaClientStreamingRequestBuilder[I, O] = new ScalaClientStreamingRequestBuilder[I, O](descriptor, channel, defaultOptions, settings, headers) + + override def setDeadline(deadline: Duration): ScalaClientStreamingRequestBuilder[I, O] = + new ScalaClientStreamingRequestBuilder[I, O]( + descriptor, + channel, + if (!deadline.isFinite) defaultOptions.withDeadline(null) + else defaultOptions.withDeadlineAfter(deadline.toMillis, TimeUnit.MILLISECONDS), + settings, + headers) } /** @@ -195,6 +224,15 @@ final class JavaClientStreamingRequestBuilder[I, O]( override def withHeaders(headers: MetadataImpl): JavaClientStreamingRequestBuilder[I, O] = new JavaClientStreamingRequestBuilder[I, O](descriptor, channel, defaultOptions, settings, headers) + + override def setDeadline(deadline: JDuration): JavaClientStreamingRequestBuilder[I, O] = + new JavaClientStreamingRequestBuilder[I, O]( + descriptor, + channel, + if (deadline == null) defaultOptions.withDeadline(null) + else defaultOptions.withDeadlineAfter(deadline.toMillis, TimeUnit.MILLISECONDS), + settings, + headers) } /** @@ -242,6 +280,15 @@ final class ScalaServerStreamingRequestBuilder[I, O]( override def withHeaders(headers: MetadataImpl): ScalaServerStreamingRequestBuilder[I, O] = new ScalaServerStreamingRequestBuilder[I, O](descriptor, channel, defaultOptions, settings, headers) + + override def setDeadline(deadline: Duration): ScalaServerStreamingRequestBuilder[I, O] = + new ScalaServerStreamingRequestBuilder[I, O]( + descriptor, + channel, + if (!deadline.isFinite) defaultOptions.withDeadline(null) + else defaultOptions.withDeadlineAfter(deadline.toMillis, TimeUnit.MILLISECONDS), + settings, + headers) } /** @@ -285,6 +332,15 @@ final class JavaServerStreamingRequestBuilder[I, O]( override def withHeaders(headers: MetadataImpl): JavaServerStreamingRequestBuilder[I, O] = new JavaServerStreamingRequestBuilder[I, O](descriptor, channel, defaultOptions, settings, headers) + + override def setDeadline(deadline: JDuration): JavaServerStreamingRequestBuilder[I, O] = + new JavaServerStreamingRequestBuilder[I, O]( + descriptor, + channel, + if (deadline == null) defaultOptions.withDeadline(null) + else defaultOptions.withDeadlineAfter(deadline.toMillis, TimeUnit.MILLISECONDS), + settings, + headers) } /** @@ -333,6 +389,15 @@ final class ScalaBidirectionalStreamingRequestBuilder[I, O]( override def withHeaders(headers: MetadataImpl): ScalaBidirectionalStreamingRequestBuilder[I, O] = new ScalaBidirectionalStreamingRequestBuilder[I, O](descriptor, channel, defaultOptions, settings, headers) + + override def setDeadline(deadline: Duration): ScalaBidirectionalStreamingRequestBuilder[I, O] = + new ScalaBidirectionalStreamingRequestBuilder[I, O]( + descriptor, + channel, + if (!deadline.isFinite) defaultOptions.withDeadline(null) + else defaultOptions.withDeadlineAfter(deadline.toMillis, TimeUnit.MILLISECONDS), + settings, + headers) } /** @@ -377,6 +442,15 @@ final class JavaBidirectionalStreamingRequestBuilder[I, O]( override def withHeaders(headers: MetadataImpl): JavaBidirectionalStreamingRequestBuilder[I, O] = new JavaBidirectionalStreamingRequestBuilder[I, O](descriptor, channel, defaultOptions, settings, headers) + + override def setDeadline(deadline: JDuration): JavaBidirectionalStreamingRequestBuilder[I, O] = + new JavaBidirectionalStreamingRequestBuilder[I, O]( + descriptor, + channel, + if (deadline == null) defaultOptions.withDeadline(null) + else defaultOptions.withDeadlineAfter(deadline.toMillis, TimeUnit.MILLISECONDS), + settings, + headers) } /** diff --git a/runtime/src/main/scala/akka/grpc/javadsl/RequestBuilder.scala b/runtime/src/main/scala/akka/grpc/javadsl/RequestBuilder.scala index dfcc9f7d6..061010182 100644 --- a/runtime/src/main/scala/akka/grpc/javadsl/RequestBuilder.scala +++ b/runtime/src/main/scala/akka/grpc/javadsl/RequestBuilder.scala @@ -4,6 +4,7 @@ package akka.grpc.javadsl +import java.time.Duration import java.util.concurrent.CompletionStage import akka.NotUsed @@ -48,6 +49,12 @@ trait SingleResponseRequestBuilder[Req, Res] { * Invoke the gRPC method with the additional metadata added and provide access to response metadata */ def invokeWithMetadata(request: Req): CompletionStage[GrpcSingleResponse[Res]] + + /** + * Set the deadline for this call + * @return A new request builder, that will use the supplied deadline when invoked + */ + def setDeadline(deadline: Duration): SingleResponseRequestBuilder[Req, Res] } /** @@ -86,4 +93,10 @@ trait StreamResponseRequestBuilder[Req, Res] { * Invoke the gRPC method with the additional metadata added and provide access to response metadata */ def invokeWithMetadata(request: Req): Source[Res, CompletionStage[GrpcResponseMetadata]] + + /** + * Set the deadline for this call + * @return A new request builder, that will use the supplied deadline when invoked + */ + def setDeadline(deadline: Duration): StreamResponseRequestBuilder[Req, Res] } diff --git a/runtime/src/main/scala/akka/grpc/scaladsl/RequestBuilder.scala b/runtime/src/main/scala/akka/grpc/scaladsl/RequestBuilder.scala index 22debc4dc..1f84fb25f 100644 --- a/runtime/src/main/scala/akka/grpc/scaladsl/RequestBuilder.scala +++ b/runtime/src/main/scala/akka/grpc/scaladsl/RequestBuilder.scala @@ -11,6 +11,7 @@ import akka.stream.scaladsl.Source import akka.util.ByteString import scala.concurrent.Future +import scala.concurrent.duration.Duration /** * Request builder for requests providing per call specific metadata capabilities in @@ -48,6 +49,12 @@ trait SingleResponseRequestBuilder[Req, Res] { * Invoke the gRPC method with the additional metadata added and provide access to response metadata */ def invokeWithMetadata(request: Req): Future[GrpcSingleResponse[Res]] + + /** + * Set the deadline for this call + * @return A new request builder, that will use the supplied deadline when invoked + */ + def setDeadline(deadline: Duration): SingleResponseRequestBuilder[Req, Res] } /** @@ -88,4 +95,10 @@ trait StreamResponseRequestBuilder[Req, Res] { * Invoke the gRPC method with the additional metadata added and provide access to response metadata */ def invokeWithMetadata(request: Req): Source[Res, Future[GrpcResponseMetadata]] + + /** + * Set the deadline for this call + * @return A new request builder, that will use the supplied deadline when invoked + */ + def setDeadline(deadline: Duration): StreamResponseRequestBuilder[Req, Res] }