raboof/pekko
1
0
Fork 0
Code Issues Pull requests Projects Releases 6 Packages Wiki Activity Actions 11

BlockingIODispatcher actually not needed #27632

Browse source
This commit is contained in:
Johan Andrén 2019-09-26 15:52:18 +02:00 • committed by GitHub
parent 119e832a0c
commit 37a76aa9a0
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
5 changed files with 3 additions and 17 deletions
Download patch file Download diff file Expand all files Collapse all files

2
akka-docs/src/main/paradox/project/migration-guide-2.5.x-2.6.x.md
View file

@ -596,7 +596,7 @@ used for individual streams when they are materialized.
| `streamRefSettings.demandRedeliveryInterval` | `StreamRefAttributes.demandRedeliveryInterval` | `akka.stream.materializer.stream-ref.demand-redelivery-interval` | | `streamRefSettings.demandRedeliveryInterval` | `StreamRefAttributes.demandRedeliveryInterval` | `akka.stream.materializer.stream-ref.demand-redelivery-interval` |
| `streamRefSettings.subscriptionTimeout` | `StreamRefAttributes.subscriptionTimeout` | `akka.stream.materializer.stream-ref.subscription-timeout` | | `streamRefSettings.subscriptionTimeout` | `StreamRefAttributes.subscriptionTimeout` | `akka.stream.materializer.stream-ref.subscription-timeout` |
| `streamRefSettings.finalTerminationSignalDeadline` | `StreamRefAttributes.finalTerminationSignalDeadline` | `akka.stream.materializer.stream-ref.final-termination-signal-deadline` | | `streamRefSettings.finalTerminationSignalDeadline` | `StreamRefAttributes.finalTerminationSignalDeadline` | `akka.stream.materializer.stream-ref.final-termination-signal-deadline` |
| `blockingIoDispatcher` | `ActorAttributes.blockingIoDispatcher` | `akka.stream.materializer.blocking-io-dispatcher` | | `blockingIoDispatcher` | na | `akka.stream.materializer.blocking-io-dispatcher` |
| `subscriptionTimeoutSettings.mode` | `ActorAttributes.streamSubscriptionTimeoutMode` | `akka.stream.materializer.subscription-timeout.mode` | | `subscriptionTimeoutSettings.mode` | `ActorAttributes.streamSubscriptionTimeoutMode` | `akka.stream.materializer.subscription-timeout.mode` |
| `subscriptionTimeoutSettings.timeout` | `ActorAttributes.streamSubscriptionTimeout` | `akka.stream.materializer.subscription-timeout.timeout` | | `subscriptionTimeoutSettings.timeout` | `ActorAttributes.streamSubscriptionTimeout` | `akka.stream.materializer.subscription-timeout.timeout` |

3
akka-stream/src/main/scala/akka/stream/ActorMaterializer.scala
View file

@ -779,8 +779,7 @@ final class ActorMaterializerSettings @InternalApi private (
ActorAttributes.OutputBurstLimit(outputBurstLimit) :: ActorAttributes.OutputBurstLimit(outputBurstLimit) ::
ActorAttributes.FuzzingMode(fuzzingMode) :: ActorAttributes.FuzzingMode(fuzzingMode) ::
ActorAttributes.MaxFixedBufferSize(maxFixedBufferSize) :: ActorAttributes.MaxFixedBufferSize(maxFixedBufferSize) ::
ActorAttributes.SyncProcessingLimit(syncProcessingLimit) :: ActorAttributes.SyncProcessingLimit(syncProcessingLimit) :: Nil)
ActorAttributes.BlockingIoDispatcher(blockingIoDispatcher) :: Nil)
override def toString: String = override def toString: String =
s"ActorMaterializerSettings($initialInputBufferSize,$maxInputBufferSize," + s"ActorMaterializerSettings($initialInputBufferSize,$maxInputBufferSize," +

6
akka-stream/src/main/scala/akka/stream/Attributes.scala
View file

@ -588,12 +588,6 @@ object ActorAttributes {
def syncProcessingLimit(limit: Int): Attributes = def syncProcessingLimit(limit: Int): Attributes =
Attributes(SyncProcessingLimit(limit)) Attributes(SyncProcessingLimit(limit))
/**
* FIXME Is this really needed anymore now that we have indirect dispatcher config?
*/
final case class BlockingIoDispatcher(dispatcher: String) extends MandatoryAttribute
def blockingIoDispatcher(dispatcher: String): Attributes =
Attributes(BlockingIoDispatcher(dispatcher))
} }
/** /**

5
akka-stream/src/main/scala/akka/stream/impl/ActorMaterializerImpl.scala
View file

@ -57,11 +57,8 @@ import scala.concurrent.Future
@InternalApi private[akka] override def actorOf(context: MaterializationContext, props: Props): ActorRef = { @InternalApi private[akka] override def actorOf(context: MaterializationContext, props: Props): ActorRef = {
val effectiveProps = props.dispatcher match { val effectiveProps = props.dispatcher match {
case Dispatchers.DefaultDispatcherId => case Dispatchers.DefaultDispatcherId =>
// the caller said to use the default dispatcher, but that can been trumped by the dispatcher attribute
props.withDispatcher(context.effectiveAttributes.mandatoryAttribute[ActorAttributes.Dispatcher].dispatcher) props.withDispatcher(context.effectiveAttributes.mandatoryAttribute[ActorAttributes.Dispatcher].dispatcher)
case ActorAttributes.IODispatcher.dispatcher =>
// this one is actually not a dispatcher but a relative config key pointing containing the actual dispatcher name
val actual = context.effectiveAttributes.mandatoryAttribute[ActorAttributes.BlockingIoDispatcher].dispatcher
props.withDispatcher(actual)
case _ => props case _ => props
} }

4
akka-stream/src/main/scala/akka/stream/impl/PhasedFusingActorMaterializer.scala
View file

@ -625,10 +625,6 @@ private final case class SavedIslandData(
val effectiveProps = props.dispatcher match { val effectiveProps = props.dispatcher match {
case Dispatchers.DefaultDispatcherId => case Dispatchers.DefaultDispatcherId =>
props.withDispatcher(context.effectiveAttributes.mandatoryAttribute[ActorAttributes.Dispatcher].dispatcher) props.withDispatcher(context.effectiveAttributes.mandatoryAttribute[ActorAttributes.Dispatcher].dispatcher)
case ActorAttributes.IODispatcher.dispatcher =>
// this one is actually not a dispatcher but a relative config key pointing containing the actual dispatcher name
// FIXME go via attributes here,or something
props.withDispatcher(settings.blockingIoDispatcher)
case _ => props case _ => props
} }