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

Wip 22430 make tlsspec pass again drewhk (#22482)

Browse source 
* #22430 Fix TLSSpec

* Remove unnecessary PushAttributes

* Almost fixing it...
This commit is contained in:
drewhk 2017-03-08 10:30:00 +01:00 • committed by GitHub
parent c028b550f2
commit c3c5065444
3 changed files with 64 additions and 36 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -112,7 +112,7 @@ private final case class ForwardWire(
override def toString: String = s"ForwardWire(islandId = $islandGlobalOffset, from = $from, toGlobal = $toGlobalOffset, phase = $phase)"
}
private final case class SavedIslandData(islandGlobalOffset: Int, skippedSlots: Int, phase: PhaseIsland[Any])
private final case class SavedIslandData(islandGlobalOffset: Int, lastVisitedOffset: Int, skippedSlots: Int, phase: PhaseIsland[Any])
class IslandTracking(
val phases: Map[IslandTag, Phase[Any]],
@ -147,7 +147,7 @@ class IslandTracking(
def getCurrentPhase: PhaseIsland[Any] = currentPhase
def getCurrentOffset: Int = currentGlobalOffset
private def completeSegment(): Int = {
private def completeSegment(): Unit = {
val length = currentGlobalOffset - currentSegmentGlobalOffset
if (activePhases eq null) {
@ -172,26 +172,25 @@ class IslandTracking(
} else {
if (Debug) println(s"Skipped zero length segment")
}
length
}
def enterIsland(tag: IslandTag, attributes: Attributes): Unit = {
completeSegment()
val previousPhase = currentPhase
val previousIslandOffset = currentIslandGlobalOffset
islandStateStack.add(SavedIslandData(previousIslandOffset, currentGlobalOffset, currentIslandSkippetSlots, previousPhase))
val effectiveSettings = materializer.effectiveSettings(attributes)
currentPhase = phases(tag)(effectiveSettings, materializer, nextIslandName())
activePhases.add(currentPhase)
if (Debug) println(s"Entering island starting at offset = $currentIslandGlobalOffset phase = $currentPhase")
// Resolve the phase to be used to materialize this island
currentIslandGlobalOffset = currentGlobalOffset
// The base offset of this segment is the current global offset
currentSegmentGlobalOffset = currentGlobalOffset
islandStateStack.add(SavedIslandData(previousIslandOffset, currentIslandSkippetSlots, previousPhase))
currentIslandSkippetSlots = 0
if (Debug) println(s"Entering island starting at offset = $currentIslandGlobalOffset phase = $currentPhase")
}
def exitIsland(): Unit = {
@ -204,7 +203,7 @@ class IslandTracking(
// We restore data for the island
currentIslandGlobalOffset = parentIsland.islandGlobalOffset
currentPhase = parentIsland.phase
currentIslandSkippetSlots = parentIsland.skippedSlots + previousSegmentLength
currentIslandSkippetSlots = parentIsland.skippedSlots + (currentGlobalOffset - parentIsland.lastVisitedOffset)
if (Debug) println(s"Exited to island starting at offset = $currentIslandGlobalOffset phase = $currentPhase")
}
@ -253,7 +252,6 @@ class IslandTracking(
}
def wireOut(out: OutPort, absoluteOffset: Int, logic: Any): Unit = {
// TODO: forward wires
if (Debug) println(s" wiring $out to absolute = $absoluteOffset")
// First check if we are wiring backwards. This is important since we can only do resolution for backward wires.
@ -760,10 +758,10 @@ final class TlsModulePhase(settings: ActorMaterializerSettings, materializer: Ph
var publishers: Vector[ActorPublisher[Any]] = _
def materializeAtomic(mod: AtomicModule[Shape, Any], attributes: Attributes): (NotUsed, Any) = {
println(mod)
val tls = mod.asInstanceOf[TlsModule]
val props = TLSActor.props(settings, tls.createSSLEngine, tls.verifySession, tls.closing)
val props =
TLSActor.props(settings, tls.createSSLEngine, tls.verifySession, tls.closing).withDispatcher(settings.dispatcher)
tlsActor = materializer.actorOf(props, islandName)
def factory(id: Int) = new ActorPublisher[Any](tlsActor) {
override val wakeUpMsg = FanOut.SubstreamSubscribePending(id)
@ -779,7 +777,7 @@ final class TlsModulePhase(settings: ActorMaterializerSettings, materializer: Ph
publishers(out.id)
def takePublisher(slot: Int, publisher: Publisher[Any]): Unit =
publisher.subscribe(FanIn.SubInput[Any](tlsActor, slot))
publisher.subscribe(FanIn.SubInput[Any](tlsActor, 1 - slot))
def onIslandReady(): Unit = ()
}