Skip to content

Commit 18747bd

Browse files
committed
Address comment
1 parent a71d296 commit 18747bd

File tree

1 file changed

+4
-32
lines changed

1 file changed

+4
-32
lines changed

stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowMapWithResourceSpec.scala

Lines changed: 4 additions & 32 deletions
Original file line numberDiff line numberDiff line change
@@ -547,35 +547,6 @@ class FlowMapWithResourceSpec extends StreamSpec(UnboundedMailboxConfig) {
547547
closedCounter.get shouldBe 1
548548
}
549549

550-
"continue with autoCloseable when Strategy is Resume and exception happened on map" in {
551-
val closedCounter = new AtomicInteger(0)
552-
val create = () =>
553-
new AutoCloseable {
554-
override def close(): Unit = closedCounter.incrementAndGet()
555-
}
556-
val p = Source
557-
.fromIterator(() => (0 to 50).iterator)
558-
.mapWithResource(create, (_: AutoCloseable, elem) => elem)
559-
.map(elem => {
560-
if (elem == 10) throw TE("") else elem
561-
})
562-
.withAttributes(supervisionStrategy(resumingDecider))
563-
.runWith(Sink.asPublisher(false))
564-
val c = TestSubscriber.manualProbe[Int]()
565-
566-
p.subscribe(c)
567-
val sub = c.expectSubscription()
568-
569-
(0 to 48).foreach(i => {
570-
sub.request(1)
571-
c.expectNext() should ===(if (i < 10) i else i + 1)
572-
})
573-
sub.request(1)
574-
c.expectNext(50)
575-
c.expectComplete()
576-
closedCounter.get shouldBe 1
577-
}
578-
579550
"close and open stream with autocloseable again when Strategy is Restart" in {
580551
val closedCounter = new AtomicInteger(0)
581552
val create = () =>
@@ -586,7 +557,7 @@ class FlowMapWithResourceSpec extends StreamSpec(UnboundedMailboxConfig) {
586557
.fromIterator(() => (0 to 50).iterator)
587558
.mapWithResource(create,
588559
(_: AutoCloseable, elem) => {
589-
if (elem == 10) throw TE("") else elem
560+
if (elem == 10 || elem == 20) throw TE("") else elem
590561
})
591562
.withAttributes(supervisionStrategy(restartingDecider))
592563
.runWith(Sink.asPublisher(false))
@@ -595,9 +566,10 @@ class FlowMapWithResourceSpec extends StreamSpec(UnboundedMailboxConfig) {
595566
p.subscribe(c)
596567
val sub = c.expectSubscription()
597568

598-
(0 to 18).foreach(i => {
569+
(0 to 30).filter(i => i != 10 && i != 20).foreach(i => {
599570
sub.request(1)
600-
c.expectNext() should ===(if (i < 10) i else i + 1)
571+
c.expectNext() shouldBe i
572+
closedCounter.get should ===(if (i < 10) 0 else if (i < 20) 1 else 2)
601573
})
602574
sub.cancel()
603575
}

0 commit comments

Comments
 (0)