@@ -1906,7 +1906,7 @@ class AsyncApiTest {
19061906
19071907 @Test
19081908 void testGroovyPromiseIsCompletedExceptionally() {
1909- def cf = new java.util.concurrent. CompletableFuture<>
1909+ def cf = new CompletableFuture<> ()
19101910 def promise = GroovyPromise . of(cf)
19111911 assert ! promise. isCompletedExceptionally()
19121912 cf. completeExceptionally(new RuntimeException ())
@@ -1919,4 +1919,119 @@ class AsyncApiTest {
19191919 new GroovyPromise (null )
19201920 }
19211921 }
1922+
1923+ // ================================================================
1924+ // AsyncStreamGenerator: complete()/error() robustness under interrupt
1925+ // ================================================================
1926+
1927+ /**
1928+ * Verifies that when a producer's complete() signal is interrupted and
1929+ * the best-effort offer() fails (no consumer waiting), the generator
1930+ * force-closes so a subsequent moveNext() returns false instead of
1931+ * blocking indefinitely.
1932+ */
1933+ @Test
1934+ void testGeneratorCompleteForceClosesOnOfferFailure() {
1935+ def gen = new AsyncStreamGenerator<Integer > ()
1936+ def producerThread = Thread . currentThread()
1937+ gen. attachProducer(producerThread)
1938+
1939+ // Interrupt the current thread so that queue.put(DONE) inside
1940+ // complete() throws InterruptedException. Since no consumer is
1941+ // blocked in take(), the non-blocking offer(DONE) will also fail,
1942+ // triggering the force-close path.
1943+ producerThread. interrupt()
1944+ gen. complete()
1945+
1946+ // Clear the interrupt flag set by the force-close path
1947+ Thread . interrupted()
1948+
1949+ gen. detachProducer(producerThread)
1950+
1951+ // Consumer should see a cleanly closed stream — not block forever
1952+ def result = gen. moveNext()
1953+ assert ! AsyncSupport . await(result) : " moveNext() should return false after force-close"
1954+ }
1955+
1956+ /**
1957+ * Verifies that when a producer's error() signal is interrupted and
1958+ * the best-effort offer() fails, the generator force-closes so the
1959+ * consumer does not hang.
1960+ */
1961+ @Test
1962+ void testGeneratorErrorForceClosesOnOfferFailure() {
1963+ def gen = new AsyncStreamGenerator<Integer > ()
1964+ def producerThread = Thread . currentThread()
1965+ gen. attachProducer(producerThread)
1966+
1967+ producerThread. interrupt()
1968+ gen. error(new RuntimeException (" test error"
1969+
1970+ Thread . interrupted()
1971+ gen. detachProducer(producerThread)
1972+
1973+ def result = gen. moveNext()
1974+ assert ! AsyncSupport . await(result) : " moveNext() should return false after force-close"
1975+ }
1976+
1977+ /**
1978+ * Verifies that complete() is a no-op when the stream is already closed.
1979+ */
1980+ @Test
1981+ void testGeneratorCompleteAfterClose() {
1982+ def gen = new AsyncStreamGenerator<Integer > ()
1983+ gen. close()
1984+ // Should not throw or block
1985+ gen. complete()
1986+ def result = gen. moveNext()
1987+ assert ! AsyncSupport . await(result)
1988+ }
1989+
1990+ /**
1991+ * Verifies that error() is a no-op when the stream is already closed.
1992+ */
1993+ @Test
1994+ void testGeneratorErrorAfterClose() {
1995+ def gen = new AsyncStreamGenerator<Integer > ()
1996+ gen. close()
1997+ // Should not throw or block
1998+ gen. error(new RuntimeException (" ignored"
1999+ def result = gen. moveNext()
2000+ assert ! AsyncSupport . await(result)
2001+ }
2002+
2003+ /**
2004+ * Verifies that close() is idempotent — multiple calls do not throw or
2005+ * cause double-interrupt of threads.
2006+ */
2007+ @Test
2008+ void testGeneratorCloseIdempotent() {
2009+ def gen = new AsyncStreamGenerator<Integer > ()
2010+ gen. close()
2011+ gen. close()
2012+ gen. close()
2013+ // All should be no-ops; moveNext should still return false
2014+ assert ! AsyncSupport . await(gen. moveNext())
2015+ }
2016+
2017+ /**
2018+ * Verifies that attachProducer on an already-closed stream immediately
2019+ * interrupts the producer thread, allowing the generator body to exit.
2020+ */
2021+ @Test
2022+ void testAttachProducerOnClosedStreamInterrupts() {
2023+ def gen = new AsyncStreamGenerator<Integer > ()
2024+ gen. close()
2025+
2026+ def interrupted = new AtomicBoolean (false )
2027+ def latch = new CountDownLatch (1 )
2028+ def t = new Thread ({
2029+ gen. attachProducer(Thread . currentThread())
2030+ interrupted. set(Thread . currentThread(). isInterrupted())
2031+ latch. countDown()
2032+ })
2033+ t. start()
2034+ latch. await(5 , TimeUnit . SECONDS )
2035+ assert interrupted. get() : " Producer should be interrupted when attached to a closed stream"
2036+ }
19222037}
0 commit comments