more refactoring (#13)

motivation: make lambda execution sequence easier to reason about

changes:
* add `peekFailure` extension to `EventLoopFuture` to handle error without side effects
* update process chain to make use of `peekFailure`

Author: tomerd

Sources/SwiftAwsLambda/LambdaRunner.swift

@@ -36,45 +36,32 @@ internal final class LambdaRunner {
     func initialize(logger: Logger) -> EventLoopFuture<Void> {
         logger.info("initializing lambda")
         // We need to use `flatMap` instead of `whenFailure` to ensure we complete reporting the result before stopping.
-        return self.lambdaHandler.initialize(eventLoop: self.eventLoop, lifecycleId: self.lifecycleId).flatMapError { error in
-            self.runtimeClient.reportInitializationError(logger: logger, error: error).flatMapResult { result -> Result<Void, Error> in
-                if case .failure(let reportingError) = result {
-                    // We're going to bail out because the init failed, so there's not a lot we can do other than log
-                    // that we couldn't report this error back to the runtime.
-                    logger.error("failed reporting initialization error to lambda runtime engine: \(reportingError)")
-                }
-                // Always return the init error
-                return .failure(error)
+        return self.lambdaHandler.initialize(eventLoop: self.eventLoop, lifecycleId: self.lifecycleId).peekError { error in
+            self.runtimeClient.reportInitializationError(logger: logger, error: error).peekError { reportingError in
+                // We're going to bail out because the init failed, so there's not a lot we can do other than log
+                // that we couldn't report this error back to the runtime.
+                logger.error("failed reporting initialization error to lambda runtime engine: \(reportingError)")
             }
         }
     }
 
     func run(logger: Logger) -> EventLoopFuture<Void> {
         logger.info("lambda invocation sequence starting")
         // 1. request work from lambda runtime engine
-        return self.runtimeClient.requestWork(logger: logger).flatMap { workRequestResult in
-            switch workRequestResult {
-            case .failure(let error):
-                logger.error("could not fetch work from lambda runtime engine: \(error)")
-                return self.eventLoop.makeFailedFuture(error)
-            case .success(let context, let payload):
-                // 2. send work to handler
-                logger.info("sending work to lambda handler \(self.lambdaHandler)")
-                return self.lambdaHandler.handle(eventLoop: self.eventLoop, lifecycleId: self.lifecycleId, context: context, payload: payload).flatMap { lambdaResult in
-                    // 3. report results to runtime engine
-                    self.runtimeClient.reportResults(logger: logger, context: context, result: lambdaResult).flatMap { postResultsResult in
-                        switch postResultsResult {
-                        case .failure(let error):
-                            logger.error("failed reporting results to lambda runtime engine: \(error)")
-                            return self.eventLoop.makeFailedFuture(error)
-                        case .success():
-                            // we are done!
-                            logger.info("lambda invocation sequence completed successfully")
-                            return self.eventLoop.makeSucceededFuture(())
-                        }
-                    }
-                }
-            }
+        return self.runtimeClient.requestWork(logger: logger).peekError { error in
+            logger.error("could not fetch work from lambda runtime engine: \(error)")
+        }.flatMap { context, payload in
+            // 2. send work to handler
+            logger.info("sending work to lambda handler \(self.lambdaHandler)")
+            return self.lambdaHandler.handle(eventLoop: self.eventLoop, lifecycleId: self.lifecycleId, context: context, payload: payload).map { (context, $0) }
+        }.flatMap { context, result in
+            // 3. report results to runtime engine
+            self.runtimeClient.reportResults(logger: logger, context: context, result: result)
+        }.peekError { error in
+            logger.error("failed reporting results to lambda runtime engine: \(error)")
+        }.always { result in
+            // we are done!
+            logger.info("lambda invocation sequence completed \(result.successful ? "successfully" : "with failure")")
         }
     }
 }
@@ -100,3 +87,36 @@ private extension LambdaHandler {
         return promise.futureResult
     }
 }
+
+// TODO: move to nio?
+private extension EventLoopFuture {
+    // callback does not have side effects, failing with original result
+    func peekError(_ callback: @escaping (Error) -> Void) -> EventLoopFuture<Value> {
+        return self.flatMapError { error in
+            callback(error)
+            return self
+        }
+    }
+
+    // callback does not have side effects, failing with original result
+    func peekError(_ callback: @escaping (Error) -> EventLoopFuture<Void>) -> EventLoopFuture<Value> {
+        return self.flatMapError { error in
+            let promise = self.eventLoop.makePromise(of: Value.self)
+            callback(error).whenComplete { _ in
+                promise.completeWith(self)
+            }
+            return promise.futureResult
+        }
+    }
+}
+
+private extension Result {
+    var successful: Bool {
+        switch self {
+        case .success:
+            return true
+        default:
+            return false
+        }
+    }
+}

Sources/SwiftAwsLambda/LambdaRuntimeClient.swift

@@ -35,24 +35,26 @@ internal class LambdaRuntimeClient {
         self.allocator = ByteBufferAllocator()
     }
 
-    func requestWork(logger: Logger) -> EventLoopFuture<RequestWorkResult> {
+    /// Requests work from the Runtime Engine.
+    func requestWork(logger: Logger) -> EventLoopFuture<(LambdaContext, [UInt8])> {
         let url = self.baseUrl + Consts.invokationURLPrefix + Consts.requestWorkURLSuffix
         logger.info("requesting work from lambda runtime engine using \(url)")
-        return self.httpClient.get(url: url).map { response in
+        return self.httpClient.get(url: url).flatMapThrowing { response in
             guard response.status == .ok else {
-                return .failure(.badStatusCode(response.status))
+                throw LambdaRuntimeClientError.badStatusCode(response.status)
             }
             guard let payload = response.readWholeBody() else {
-                return .failure(.noBody)
+                throw LambdaRuntimeClientError.noBody
             }
             guard let context = LambdaContext(logger: logger, response: response) else {
-                return .failure(.noContext)
+                throw LambdaRuntimeClientError.noContext
             }
-            return .success((context, payload))
+            return (context, payload)
         }
     }
 
-    func reportResults(logger: Logger, context: LambdaContext, result: LambdaResult) -> EventLoopFuture<PostResultsResult> {
+    /// Reports a result to the Runtime Engine.
+    func reportResults(logger: Logger, context: LambdaContext, result: LambdaResult) -> EventLoopFuture<Void> {
         var url = self.baseUrl + Consts.invokationURLPrefix + "/" + context.requestId
         var body: ByteBuffer
         switch result {
@@ -66,41 +68,43 @@ internal class LambdaRuntimeClient {
             let error = ErrorResponse(errorType: "FunctionError", errorMessage: "\(error)")
             switch error.toJson() {
             case .failure(let jsonError):
-                return self.eventLoop.makeSucceededFuture(.failure(.json(jsonError)))
+                return self.eventLoop.makeFailedFuture(LambdaRuntimeClientError.json(jsonError))
             case .success(let json):
                 body = self.allocator.buffer(capacity: json.utf8.count)
                 body.writeString(json)
             }
         }
         logger.info("reporting results to lambda runtime engine using \(url)")
-        return self.httpClient.post(url: url, body: body).map { response in
-            response.status != .accepted ? .failure(.badStatusCode(response.status)) : .success(())
+        return self.httpClient.post(url: url, body: body).flatMapThrowing { response in
+            guard response.status == .accepted else {
+                throw LambdaRuntimeClientError.badStatusCode(response.status)
+            }
+            return ()
         }
     }
 
     /// Reports an initialization error to the Runtime Engine.
-    func reportInitializationError(logger: Logger, error: Error) -> EventLoopFuture<PostInitializationErrorResult> {
+    func reportInitializationError(logger: Logger, error: Error) -> EventLoopFuture<Void> {
         let url = self.baseUrl + Consts.postInitErrorURL
         let errorResponse = ErrorResponse(errorType: "InitializationError", errorMessage: "\(error)")
         var body: ByteBuffer
         switch errorResponse.toJson() {
         case .failure(let jsonError):
-            return self.eventLoop.makeSucceededFuture(.failure(.json(jsonError)))
+            return self.eventLoop.makeFailedFuture(LambdaRuntimeClientError.json(jsonError))
         case .success(let json):
             body = self.allocator.buffer(capacity: json.utf8.count)
             body.writeString(json)
             logger.info("reporting initialization error to lambda runtime engine using \(url)")
-            return self.httpClient.post(url: url, body: body).map { response in
-                response.status != .accepted ? .failure(.badStatusCode(response.status)) : .success(())
+            return self.httpClient.post(url: url, body: body).flatMapThrowing { response in
+                guard response.status == .accepted else {
+                    throw LambdaRuntimeClientError.badStatusCode(response.status)
+                }
+                return ()
             }
         }
     }
 }
 
-internal typealias RequestWorkResult = Result<(LambdaContext, [UInt8]), LambdaRuntimeClientError>
-internal typealias PostResultsResult = Result<Void, LambdaRuntimeClientError>
-internal typealias PostInitializationErrorResult = Result<Void, LambdaRuntimeClientError>
-
 internal enum LambdaRuntimeClientError: Error, Equatable {
     case badStatusCode(HTTPResponseStatus)
     case noBody

Tests/SwiftAwsLambdaTests/LambdaTest.swift

@@ -84,7 +84,7 @@ class LambdaTest: XCTestCase {
             kill(getpid(), signal.rawValue)
         }
         let result = try future.wait()
-        XCTAssertGreaterThan(result, 0, "should have stopped before any reuqetsst made")
+        XCTAssertGreaterThan(result, 0, "should have stopped before any request made")
         XCTAssertLessThan(result, max, "should have stopped before \(max)")
         try server.stop().wait()
     }