Skip to content

Commit 992fbb3

Browse files
sdelamosdelamo
authored
fix: handle RequestStreamHandler in AbstractMicronautLambdaRuntime (#1997)
* fix: handle RequestStreamHandler in AbstractMicronautLambdaRuntime Close #1871 * add tests for handler validation * sonar * sonar: Remove usage of generic wildcard type. * sonar reduce complexity * fix test * increase test coverage * sonar: Provide the parametrized type for this generic. * checkstyle: document T
1 parent 51d9fc7 commit 992fbb3

9 files changed

+413
-55
lines changed

‎function-aws-custom-runtime/src/main/java/io/micronaut/function/aws/runtime/AbstractMicronautLambdaRuntime.java

+133-55
Original file line numberDiff line numberDiff line change
@@ -54,6 +54,9 @@
5454
import io.micronaut.json.JsonMapper;
5555
import io.micronaut.logging.LogLevel;
5656

57+
import java.io.InputStream;
58+
import java.io.ByteArrayInputStream;
59+
import java.io.ByteArrayOutputStream;
5760
import java.io.Closeable;
5861
import java.io.IOException;
5962
import java.io.PrintWriter;
@@ -354,6 +357,22 @@ protected HandlerRequestType createHandlerRequest(RequestType request) throws IO
354357
return null;
355358
}
356359

360+
/**
361+
* Creates a GET request for the {@value #NEXT_INVOCATION_URI} endpoint.
362+
* If a bean of type {@link UserAgentProvider} exists, it adds an HTTP Header User-Agent to the request.
363+
* @param userAgentProvider UseAgent Provider
364+
* @param <T> The Http request type
365+
* @return a Mutable HTTP Request to the {@value #NEXT_INVOCATION_URI} endpoint.
366+
*/
367+
@NonNull
368+
protected <T> MutableHttpRequest<T> createNextInvocationHttpRequest(@Nullable UserAgentProvider userAgentProvider) {
369+
MutableHttpRequest<T> nextInvocationHttpRequest = HttpRequest.GET(AwsLambdaRuntimeApi.NEXT_INVOCATION_URI);
370+
if (userAgentProvider != null) {
371+
nextInvocationHttpRequest.header(USER_AGENT, userAgentProvider.userAgent());
372+
}
373+
return nextInvocationHttpRequest;
374+
}
375+
357376
/**
358377
* Starts the runtime API event loop.
359378
*
@@ -371,83 +390,142 @@ protected void startRuntimeApiEventLoop(@NonNull URL runtimeApiURL,
371390
if (applicationContext == null) {
372391
throw new ConfigurationException("Application Context is null");
373392
}
393+
UserAgentProvider userAgentProvider = applicationContext.findBean(UserAgentProvider.class).orElse(null);
374394
populateUserAgent();
375395
final DefaultHttpClientConfiguration config = new DefaultHttpClientConfiguration();
376396
config.setReadIdleTimeout(null);
377397
config.setReadTimeout(null);
378398
config.setConnectTimeout(null);
379-
final HttpClient endpointClient = applicationContext.createBean(
380-
HttpClient.class,
381-
runtimeApiURL,
382-
config);
383-
final BlockingHttpClient blockingHttpClient = endpointClient.toBlocking();
384-
try {
385-
while (loopUntil.test(runtimeApiURL)) {
386-
MutableHttpRequest<?> nextInvocationHttpRequest = HttpRequest.GET(AwsLambdaRuntimeApi.NEXT_INVOCATION_URI);
387-
applicationContext.findBean(UserAgentProvider.class)
388-
.ifPresent(userAgentProvider -> nextInvocationHttpRequest.header(USER_AGENT, userAgentProvider.userAgent()));
389-
final HttpResponse<RequestType> response = blockingHttpClient.exchange(nextInvocationHttpRequest, Argument.of(requestType));
390-
final RequestType request = response.body();
391-
if (request != null) {
392-
logn(LogLevel.DEBUG, "request body ", request);
393-
394-
HandlerRequestType handlerRequest = createHandlerRequest(request);
395-
final HttpHeaders headers = response.getHeaders();
396-
propagateTraceId(headers);
397-
398-
final Context context = new RuntimeContext(headers);
399-
final String requestId = context.getAwsRequestId();
400-
logn(LogLevel.DEBUG, "request id ", requestId, " found");
401-
try {
402-
if (StringUtils.isNotEmpty(requestId)) {
403-
log(LogLevel.TRACE, "invoking handler\n");
404-
HandlerResponseType handlerResponse = null;
405-
if (handler instanceof RequestHandler) {
406-
handlerResponse = ((RequestHandler<HandlerRequestType, HandlerResponseType>) handler).handleRequest(handlerRequest, context);
407-
}
408-
log(LogLevel.TRACE, "handler response received\n");
409-
final ResponseType functionResponse = (handlerResponse == null || handlerResponse instanceof Void) ? null : createResponse(handlerResponse);
410-
log(LogLevel.TRACE, "sending function response\n");
411-
blockingHttpClient.exchange(decorateWithUserAgent(invocationResponseRequest(requestId, functionResponse == null ? "" : functionResponse)));
412-
} else {
413-
log(LogLevel.WARN, "request id is empty\n");
414-
}
415-
416-
} catch (Throwable e) {
417-
final StringWriter sw = new StringWriter();
418-
e.printStackTrace(new PrintWriter(sw));
419-
logn(LogLevel.WARN, "Invocation with requestId [", requestId, "] failed: ", e.getMessage(), sw);
420-
try {
421-
blockingHttpClient.exchange(decorateWithUserAgent(invocationErrorRequest(requestId, e.getMessage(), null, null)));
422-
} catch (Throwable e2) {
423-
// swallow, nothing we can do...
424-
}
399+
try (HttpClient endpointClient = applicationContext.createBean(HttpClient.class, runtimeApiURL, config)) {
400+
final BlockingHttpClient blockingHttpClient = endpointClient.toBlocking();
401+
try {
402+
while (loopUntil.test(runtimeApiURL)) {
403+
MutableHttpRequest<?> nextInvocationHttpRequest = createNextInvocationHttpRequest(userAgentProvider);
404+
if (handler instanceof RequestStreamHandler) {
405+
handleInvocationForRequestStreamHandler(blockingHttpClient, nextInvocationHttpRequest);
406+
} else if (handler instanceof RequestHandler<?, ?>) {
407+
handleInvocationForRequestHandler(blockingHttpClient, nextInvocationHttpRequest);
425408
}
426409
}
427-
}
428-
} finally {
429-
if (handler instanceof Closeable closeable) {
430-
closeable.close();
431-
}
432-
if (endpointClient != null) {
433-
endpointClient.close();
410+
} finally {
411+
if (handler instanceof Closeable closeable) {
412+
closeable.close();
413+
}
434414
}
435415
}
436-
} catch (Throwable e) {
416+
} catch (Exception e) {
437417
e.printStackTrace();
438418
logn(LogLevel.ERROR, "Request loop failed with: ", e.getMessage());
439419
reportInitializationError(runtimeApiURL, e);
440420
}
441421
}
442422

423+
/**
424+
* It handles an invocation event with a handler of type {@link RequestHandler}.
425+
* @param blockingHttpClient Blocking HTTP Client
426+
* @param nextInvocationHttpRequest Next Invocation HTTP Request
427+
* @throws IOException Exception thrown while invoking the handler
428+
*/
429+
protected void handleInvocationForRequestHandler(@NonNull BlockingHttpClient blockingHttpClient,
430+
@NonNull MutableHttpRequest<?> nextInvocationHttpRequest) throws IOException {
431+
final HttpResponse<RequestType> response = blockingHttpClient.exchange(nextInvocationHttpRequest, Argument.of(requestType));
432+
final RequestType request = response.body();
433+
if (request != null && handler instanceof RequestHandler) {
434+
logn(LogLevel.DEBUG, "request body ", request);
435+
Context context = createRuntimeContext(response);
436+
final String requestId = context.getAwsRequestId();
437+
HandlerRequestType handlerRequest = createHandlerRequest(request);
438+
try {
439+
if (StringUtils.isEmpty(requestId)) {
440+
log(LogLevel.WARN, "request id is empty\n");
441+
return;
442+
}
443+
log(LogLevel.TRACE, "invoking handler\n");
444+
HandlerResponseType handlerResponse = null;
445+
handlerResponse = ((RequestHandler<HandlerRequestType, HandlerResponseType>) handler).handleRequest(handlerRequest, context);
446+
log(LogLevel.TRACE, "handler response received\n");
447+
final ResponseType functionResponse = (handlerResponse == null || handlerResponse instanceof Void) ? null : createResponse(handlerResponse);
448+
log(LogLevel.TRACE, "sending function response\n");
449+
blockingHttpClient.exchange(decorateWithUserAgent(invocationResponseRequest(requestId, functionResponse == null ? "" : functionResponse)));
450+
} catch (Exception e) {
451+
handleInvocationException(blockingHttpClient, requestId, e);
452+
}
453+
}
454+
}
455+
456+
/**
457+
*
458+
* @param blockingHttpClient Blocking HTTP Client
459+
* @param requestId AWS Request ID retried via {@link Context#getAwsRequestId()}
460+
* @param exception Execption thrown invoking the handler
461+
*/
462+
protected void handleInvocationException(@NonNull BlockingHttpClient blockingHttpClient,
463+
@NonNull String requestId,
464+
@NonNull Exception exception) {
465+
final StringWriter sw = new StringWriter();
466+
exception.printStackTrace(new PrintWriter(sw));
467+
logn(LogLevel.WARN, "Invocation with requestId [", requestId, "] failed: ", exception.getMessage(), sw);
468+
try {
469+
blockingHttpClient.exchange(decorateWithUserAgent(invocationErrorRequest(requestId, exception.getMessage(), null, null)));
470+
} catch (Exception e2) {
471+
// swallow, nothing we can do...
472+
}
473+
}
474+
475+
/**
476+
*
477+
* @param response Next Invocation Response
478+
* @return a new {@link Context} backed by a {@link RuntimeContext} populated with the HTTP Headers of the Invocation Response.
479+
*/
480+
protected Context createRuntimeContext(HttpResponse<?> response) {
481+
final HttpHeaders headers = response.getHeaders();
482+
propagateTraceId(headers);
483+
final Context context = new RuntimeContext(headers);
484+
final String requestId = context.getAwsRequestId();
485+
logn(LogLevel.DEBUG, "request id ", requestId, " found");
486+
return context;
487+
}
488+
489+
/**
490+
* It handles an invocation event with a handler of type {@link RequestStreamHandler}.
491+
* @param blockingHttpClient Blocking HTTP Client
492+
* @param nextInvocationHttpRequest Next Invocation HTTP Request
493+
*/
494+
protected void handleInvocationForRequestStreamHandler(@NonNull BlockingHttpClient blockingHttpClient,
495+
MutableHttpRequest<?> nextInvocationHttpRequest) {
496+
if (handler instanceof RequestStreamHandler requestStreamHandler) {
497+
final HttpResponse<byte[]> response = blockingHttpClient.exchange(nextInvocationHttpRequest, byte[].class);
498+
final byte[] request = response.body();
499+
if (request != null) {
500+
Context context = createRuntimeContext(response);
501+
String requestId = context.getAwsRequestId();
502+
if (StringUtils.isNotEmpty(requestId)) {
503+
try (InputStream inputStream = new ByteArrayInputStream(request)) {
504+
ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
505+
log(LogLevel.TRACE, "invoking handler\n");
506+
requestStreamHandler.handleRequest(inputStream, outputStream, context);
507+
log(LogLevel.TRACE, "handler response received\n");
508+
byte[] handlerResponse = outputStream.toByteArray();
509+
log(LogLevel.TRACE, "sending function response\n");
510+
blockingHttpClient.exchange(decorateWithUserAgent(invocationResponseRequest(requestId, handlerResponse)));
511+
} catch (Exception e) {
512+
handleInvocationException(blockingHttpClient, requestId, e);
513+
}
514+
}
515+
} else {
516+
log(LogLevel.WARN, "request id is empty\n");
517+
}
518+
}
519+
}
520+
443521
/**
444522
* If the request is {@link MutableHttpRequest} and {@link AbstractMicronautLambdaRuntime#userAgent} is not null,
445523
* it adds an HTTP Header User-Agent.
446524
* @param request HTTP Request
447525
* @return The HTTP Request decorated
448526
*/
449527
protected HttpRequest decorateWithUserAgent(HttpRequest<?> request) {
450-
if (userAgent != null && request instanceof MutableHttpRequest mutableHttpRequest) {
528+
if (userAgent != null && request instanceof MutableHttpRequest<?> mutableHttpRequest) {
451529
return mutableHttpRequest.header(USER_AGENT, userAgent);
452530
}
453531
return request;

‎function-aws-custom-runtime/src/main/java/io/micronaut/function/aws/runtime/AbstractRequestStreamHandlerMicronautLambdaRuntime.java

+43
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,43 @@
1+
/*
2+
* Copyright 2017-2024 original authors
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* https://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
package io.micronaut.function.aws.runtime;
17+
18+
import com.amazonaws.services.lambda.runtime.RequestStreamHandler;
19+
import io.micronaut.context.exceptions.ConfigurationException;
20+
21+
import java.io.InputStream;
22+
import java.io.OutputStream;
23+
24+
/**
25+
* Class that can be used as an entry point for an AWS Lambda custom runtime with a {@link io.micronaut.function.aws.MicronautRequestStreamHandler} handler.
26+
* @see <a href="https://docs.aws.amazon.com/lambda/latest/dg/runtimes-custom.html">Custom AWS Lambda runtimes</a>.
27+
*
28+
* @param <RequestType> The expected request object. This is the model class that the event JSON is de-serialized to
29+
* @param <ResponseType> The expected Lambda function response object. Responses will be written to this model object
30+
*
31+
* @author sdelamo
32+
* @since 4.2.2
33+
*/
34+
public abstract class AbstractRequestStreamHandlerMicronautLambdaRuntime<RequestType, ResponseType> extends AbstractMicronautLambdaRuntime<RequestType, ResponseType, InputStream, OutputStream> {
35+
36+
@Override
37+
protected void validateHandler() throws ConfigurationException {
38+
super.validateHandler();
39+
if (!(handler instanceof RequestStreamHandler)) {
40+
throw new ConfigurationException("handler must be of type com.amazonaws.services.lambda.runtime.RequestStreamHandler");
41+
}
42+
}
43+
}

‎function-aws-custom-runtime/src/test/groovy/io/micronaut/function/aws/runtime/AbstractMicronautLambdaRuntimeSpec.groovy

+44
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,44 @@
1+
package io.micronaut.function.aws.runtime
2+
3+
import com.amazonaws.services.lambda.runtime.events.APIGatewayProxyRequestEvent
4+
import com.amazonaws.services.lambda.runtime.events.APIGatewayProxyResponseEvent
5+
import io.micronaut.context.exceptions.ConfigurationException
6+
import io.micronaut.http.HttpHeaders
7+
import io.micronaut.http.HttpStatus
8+
import io.micronaut.http.MediaType
9+
import spock.lang.Specification
10+
11+
12+
class AbstractMicronautLambdaRuntimeSpec extends Specification {
13+
14+
void "validateHandler fails if handler is null"() {
15+
given:
16+
CustomRuntime customRuntime = new CustomRuntime()
17+
18+
when:
19+
customRuntime.validateHandler()
20+
21+
then:
22+
thrown(ConfigurationException)
23+
24+
and:
25+
null == customRuntime.createRequestHandler()
26+
27+
and:
28+
null == customRuntime.createRequestStreamHandler()
29+
30+
when:
31+
APIGatewayProxyResponseEvent responseEvent = customRuntime.respond(HttpStatus.I_AM_A_TEAPOT, "{\"foo\":\"bar\"}".getBytes(), MediaType.APPLICATION_JSON_GITHUB)
32+
then:
33+
responseEvent.isBase64Encoded
34+
418 == responseEvent.statusCode
35+
MediaType.APPLICATION_JSON_GITHUB == responseEvent.getHeaders().get(HttpHeaders.CONTENT_TYPE)
36+
new String(Base64.encoder.encode("{\"foo\":\"bar\"}".getBytes())) == responseEvent.body
37+
}
38+
39+
static class CustomRuntime extends AbstractMicronautLambdaRuntime<APIGatewayProxyRequestEvent, APIGatewayProxyResponseEvent, APIGatewayProxyRequestEvent, APIGatewayProxyResponseEvent> {
40+
}
41+
}
42+
43+
44+

‎function-aws-custom-runtime/src/test/groovy/io/micronaut/function/aws/runtime/micronautrequeststreamhandler/AbstractRequestStreamHandlerMicronautLambdaRuntimeSpec.groovy

+41
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,41 @@
1+
package io.micronaut.function.aws.runtime.micronautrequeststreamhandler
2+
3+
import com.amazonaws.services.lambda.runtime.events.APIGatewayProxyRequestEvent
4+
import com.amazonaws.services.lambda.runtime.events.APIGatewayProxyResponseEvent
5+
import io.micronaut.context.exceptions.ConfigurationException
6+
import io.micronaut.function.aws.MicronautRequestHandler
7+
import io.micronaut.function.aws.runtime.AbstractRequestStreamHandlerMicronautLambdaRuntime
8+
import spock.lang.Specification
9+
10+
class AbstractRequestStreamHandlerMicronautLambdaRuntimeSpec extends Specification {
11+
12+
void "validateHandler fails if RequestHandler"() {
13+
given:
14+
CustomRuntime customRuntime = new CustomRuntime()
15+
16+
when:
17+
customRuntime.validateHandler()
18+
19+
then:
20+
thrown(ConfigurationException)
21+
}
22+
23+
static class CustomRuntime extends AbstractRequestStreamHandlerMicronautLambdaRuntime<APIGatewayProxyRequestEvent, APIGatewayProxyResponseEvent> {
24+
25+
@Override
26+
protected Object createHandler(String... args) {
27+
new FooHandler()
28+
}
29+
}
30+
31+
static class FooHandler extends MicronautRequestHandler<APIGatewayProxyRequestEvent, APIGatewayProxyResponseEvent> {
32+
33+
@Override
34+
APIGatewayProxyResponseEvent execute(APIGatewayProxyRequestEvent input) {
35+
null
36+
}
37+
}
38+
}
39+
40+
41+

0 commit comments

Comments
 (0)