Merge branch 'sebsto/fix_584' into sebsto/multiple_continuations

Author: Sebastien Stormacq

Sources/AWSLambdaRuntime/Lambda+LocalServer.swift

@@ -272,7 +272,7 @@ internal struct LambdaHTTPServer {
 
                             // for streaming requests, push a partial head response
                             if self.isStreamingResponse(requestHead) {
-                                await self.responsePool.push(
+                                self.responsePool.push(
                                     LocalServerResponse(
                                         id: requestId,
                                         status: .ok
@@ -286,7 +286,7 @@ internal struct LambdaHTTPServer {
                             // if this is a request from a Streaming Lambda Handler,
                             // stream the response instead of buffering it
                             if self.isStreamingResponse(requestHead) {
-                                await self.responsePool.push(
+                                self.responsePool.push(
                                     LocalServerResponse(id: requestId, body: body)
                                 )
                             } else {
@@ -298,7 +298,7 @@ internal struct LambdaHTTPServer {
 
                             if self.isStreamingResponse(requestHead) {
                                 // for streaming response, send the final response
-                                await self.responsePool.push(
+                                self.responsePool.push(
                                     LocalServerResponse(id: requestId, final: true)
                                 )
                             } else {
@@ -390,7 +390,7 @@ internal struct LambdaHTTPServer {
             logger[metadataKey: "requestId"] = "\(requestId)"
 
             logger.trace("/invoke received invocation, pushing it to the pool and wait for a lambda response")
-            await self.invocationPool.push(LocalServerInvocation(requestId: requestId, request: body))
+            self.invocationPool.push(LocalServerInvocation(requestId: requestId, request: body))
 
             // wait for the lambda function to process the request
             for try await response in self.responsePool {
@@ -463,7 +463,7 @@ internal struct LambdaHTTPServer {
             }
             // enqueue the lambda function response to be served as response to the client /invoke
             logger.trace("/:requestId/response received response", metadata: ["requestId": "\(requestId)"])
-            await self.responsePool.push(
+            self.responsePool.push(
                 LocalServerResponse(
                     id: requestId,
                     status: .accepted,
@@ -494,7 +494,7 @@ internal struct LambdaHTTPServer {
             }
             // enqueue the lambda function response to be served as response to the client /invoke
             logger.trace("/:requestId/response received response", metadata: ["requestId": "\(requestId)"])
-            await self.responsePool.push(
+            self.responsePool.push(
                 LocalServerResponse(
                     id: requestId,
                     status: .internalServerError,
@@ -560,21 +560,21 @@ internal struct LambdaHTTPServer {
         typealias Element = T
 
         struct State: ~Copyable {
-            var actions: Deque<T>
+            var actionQueue: Deque<T>
             var continuationQueue: Deque<CheckedContinuation<T, any Error>>
         }
-        private let lock = Mutex<State>(State(actions: [], continuationQueue: []))
+        private let lock = Mutex<State>(State(actionQueue: [], continuationQueue: []))
 
         /// enqueue an element, or give it back immediately to the iterator if it is waiting for an element
-        public func push(_ action: T) async {
+        public func push(_ action: T) {
 
             // if the iterator is waiting for an element on `next()`, give it to it (resume the continuation)
             // otherwise, enqueue the element
             let maybeContinuation = self.lock.withLock { state -> CheckedContinuation<T, any Error>? in
                 if let nextContinuation = state.continuationQueue.popFirst() {
                     return nextContinuation
                 } else {
-                    state.actions.append(action)
+                    state.actionQueue.append(action)
                     return nil
                 }
             }
@@ -592,7 +592,7 @@ internal struct LambdaHTTPServer {
                     let (nextAction, nextContinuation) = self.lock.withLock {
                         state -> (T?, CheckedContinuation<T, any Error>) in
                         if let nextContinuation = state.continuationQueue.popFirst() {
-                            if let nextAction = state.actions.popFirst() {
+                            if let nextAction = state.actionQueue.popFirst() {
                                 return (nextAction, nextContinuation)
                             } else {
                                 // next is called when there is no action to process, but there is a continuation waiting

Tests/AWSLambdaRuntimeTests/PoolTests.swift

@@ -24,8 +24,8 @@ struct PoolTests {
         let pool = LambdaHTTPServer.Pool<String>()
 
         // Push values
-        await pool.push("first")
-        await pool.push("second")
+        pool.push("first")
+        pool.push("second")
 
         // Iterate and verify order
         var values = [String]()
@@ -53,7 +53,9 @@ struct PoolTests {
         task.cancel()
 
         // This should complete without receiving any values
-        try await task.value
+        do {
+            try await task.value
+        } catch is CancellationError {}  // this might happen depending on the order on which the cancellation is handled
     }
 
     @Test
@@ -78,7 +80,7 @@ struct PoolTests {
         try await withThrowingTaskGroup(of: Void.self) { group in
             for i in 0..<iterations {
                 group.addTask {
-                    await pool.push(i)
+                    pool.push(i)
                 }
             }
             try await group.waitForAll()
@@ -110,7 +112,7 @@ struct PoolTests {
         try await Task.sleep(nanoseconds: 100_000_000)  // 0.1 seconds
 
         // Push a value
-        await pool.push(expectedValue)
+        pool.push(expectedValue)
 
         // Wait for consumer to complete
         try await consumer.value
@@ -140,7 +142,7 @@ struct PoolTests {
             for p in 0..<producerCount {
                 group.addTask {
                     for i in 0..<messagesPerProducer {
-                        await pool.push(p * messagesPerProducer + i)
+                        pool.push(p * messagesPerProducer + i)
                     }
                 }
             }