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

re-introduce removed parameter (#31436)

Browse source 
To avoid breaking internal binary compatibility between 2.6.19 and 2.6.20

fixes #31431
This commit is contained in:
Arnout Engelen 2022-06-14 09:22:49 +02:00 • committed by GitHub
parent 2c85b86be1
commit a81ab6dc83
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
1 changed files with 9 additions and 8 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -290,7 +290,7 @@ private final case class SavedIslandData(
if (Debug)
println(s" cross island forward wiring from port ${forwardWire.from} wired to local slot = $localInSlot")
val publisher = forwardWire.phase.createPublisher(forwardWire.from, forwardWire.outStage)
currentPhase.takePublisher(localInSlot, publisher)
currentPhase.takePublisher(localInSlot, publisher, null)
}
}
@ -338,7 +338,7 @@ private final case class SavedIslandData(
} else {
if (Debug) println(s" cross-island wiring to local slot $localInSlot in target island")
val publisher = currentPhase.createPublisher(out, logic)
targetSegment.phase.takePublisher(localInSlot, publisher)
targetSegment.phase.takePublisher(localInSlot, publisher, null)
}
}
} else {
@ -659,7 +659,7 @@ private final case class SavedIslandData(
def createPublisher(out: OutPort, logic: M): Publisher[Any]
@InternalStableApi
def takePublisher(slot: Int, publisher: Publisher[Any]): Unit
def takePublisher(slot: Int, publisher: Publisher[Any], attributes: Attributes): Unit
def onIslandReady(): Unit
@ -767,7 +767,7 @@ private final case class SavedIslandData(
boundary.publisher
}
override def takePublisher(slot: Int, publisher: Publisher[Any]): Unit = {
override def takePublisher(slot: Int, publisher: Publisher[Any], attributes: Attributes): Unit = {
val connection = conn(slot)
val bufferSize = connection.inOwner.attributes.mandatoryAttribute[InputBuffer].max
val boundary =
@ -866,7 +866,7 @@ private final case class SavedIslandData(
override def createPublisher(out: OutPort, logic: Publisher[Any]): Publisher[Any] = logic
override def takePublisher(slot: Int, publisher: Publisher[Any]): Unit =
override def takePublisher(slot: Int, publisher: Publisher[Any], attributes: Attributes): Unit =
throw new UnsupportedOperationException("A Source cannot take a Publisher")
override def onIslandReady(): Unit = ()
@ -903,7 +903,7 @@ private final case class SavedIslandData(
throw new UnsupportedOperationException("A Sink cannot create a Publisher")
}
override def takePublisher(slot: Int, publisher: Publisher[Any]): Unit = {
override def takePublisher(slot: Int, publisher: Publisher[Any], attributes: Attributes): Unit = {
subscriberOrVirtualPublisher match {
case v: VirtualPublisher[_] => v.registerPublisher(publisher)
case s: Subscriber[Any] @unchecked => publisher.subscribe(s)
@ -936,7 +936,8 @@ private final case class SavedIslandData(
override def assignPort(out: OutPort, slot: Int, logic: Processor[Any, Any]): Unit = ()
override def createPublisher(out: OutPort, logic: Processor[Any, Any]): Publisher[Any] = logic
override def takePublisher(slot: Int, publisher: Publisher[Any]): Unit = publisher.subscribe(processor)
override def takePublisher(slot: Int, publisher: Publisher[Any], attributes: Attributes): Unit =
publisher.subscribe(processor)
override def onIslandReady(): Unit = ()
}
@ -978,7 +979,7 @@ private final case class SavedIslandData(
def createPublisher(out: OutPort, logic: NotUsed): Publisher[Any] =
publishers(out.id)
def takePublisher(slot: Int, publisher: Publisher[Any]): Unit =
override def takePublisher(slot: Int, publisher: Publisher[Any], attributes: Attributes): Unit =
publisher.subscribe(FanIn.SubInput[Any](tlsActor, 1 - slot))
def onIslandReady(): Unit = ()