Add support non-http lambda event passthrough

Author: yacchi

README.md

@@ -30,6 +30,23 @@ func echoReplyHandler(w http.ResponseWriter, r *http.Request) {
 }
 ```
 
+## Pass through non-http events
+
+Pass-through of non-HTTP Events has been added in v0.5.0.
+The destination path is /events by default.
+If you want to change the forwarding path or stop forwarding, please refer to the following sample for configuration.
+
+```go
+func main() {
+  log.Fatalln(adaptor.ListenAndServeWithOptions(
+    "",
+    handler,
+    // aws.WithNonHTTPEventPath("/api"),
+	// aws.WithoutNonHTTPEventPassThrough(),
+  )
+}
+```
+
 ## Features
 - AWS Lambda support
   - [x] API Gateway REST API integration
@@ -45,6 +62,7 @@ func echoReplyHandler(w http.ResponseWriter, r *http.Request) {
   - [x] Get raw request value from Context
   - [x] Lambda container image function
   - [x] API Gateway Websocket API integration (Experimental)
+  - [x] Non-HTTP event pass-through
 - AWS API Gateway utilities
   - [x] Strip stage var middleware
   - [ ] Abstract interface of RequestContext  

aws/lambda.go

@@ -13,6 +13,8 @@ import (
 	"net/http"
 )
 
+const DefaultNonHTTPEventPath = "/events"
+
 type LambdaHandlerOption func(handler *LambdaHandler)
 
 type SDKSessionProvider func() (*session.Session, error)
@@ -31,14 +33,27 @@ func WithAWSConfigProvider(sp SDKConfigProvider) LambdaHandlerOption {
 	}
 }
 
+func WithNonHTTPEventPath(path string) LambdaHandlerOption {
+	return func(handler *LambdaHandler) {
+		handler.nonHTTPEventPath = path
+	}
+}
+
+func WithoutNonHTTPEventPassThrough() LambdaHandlerOption {
+	return func(handler *LambdaHandler) {
+		handler.nonHTTPEventPath = ""
+	}
+}
+
 type LambdaHandler struct {
-	httpHandler  http.Handler
-	sessProv     SDKSessionProvider
-	sess         *session.Session
-	confProv     SDKConfigProvider
-	conf         *aws.Config
-	apiGW        APIGatewayManagementAPI
-	wsPathPrefix string
+	httpHandler      http.Handler
+	sessProv         SDKSessionProvider
+	sess             *session.Session
+	confProv         SDKConfigProvider
+	conf             *aws.Config
+	apiGW            APIGatewayManagementAPI
+	wsPathPrefix     string
+	nonHTTPEventPath string
 }
 
 func NewLambdaHandlerWithOption(h http.Handler, options []interface{}) lambda.Handler {
@@ -47,7 +62,8 @@ func NewLambdaHandlerWithOption(h http.Handler, options []interface{}) lambda.Ha
 		confProv: func(ctx context.Context) (aws.Config, error) {
 			return config.LoadDefaultConfig(ctx)
 		},
-		wsPathPrefix: DefaultWebsocketPathPrefix,
+		wsPathPrefix:     DefaultWebsocketPathPrefix,
+		nonHTTPEventPath: DefaultNonHTTPEventPath,
 	}
 
 	for _, opt := range options {
@@ -96,6 +112,19 @@ func (l *LambdaHandler) InvokeALBTargetGroup(ctx context.Context, request *event
 	return ALBTargetResponse(w, multiValue)
 }
 
+func (l *LambdaHandler) HandleNonHTTPEvent(ctx context.Context, event []byte, contentType string) ([]byte, error) {
+	if l.nonHTTPEventPath == "" {
+		return nil, fmt.Errorf("unknown lambda integration type and non-http event path is not set")
+	}
+	req, err := NewLambdaPassthroughRequest(ctx, event, l.nonHTTPEventPath, contentType)
+	if err != nil {
+		return nil, err
+	}
+	w := NewResponseWriter()
+	l.httpHandler.ServeHTTP(w, req)
+	return w.buf.Bytes(), nil
+}
+
 func (l *LambdaHandler) ProvideAPIGatewayClient(ctx context.Context, request *events.APIGatewayWebsocketProxyRequest) (client APIGatewayManagementAPI, err error) {
 	if l.apiGW != nil {
 		return l.apiGW, nil
@@ -158,46 +187,46 @@ func (l *LambdaHandler) Invoke(ctx context.Context, payload []byte) ([]byte, err
 		err     error
 	)
 
-	if err := json.Unmarshal(payload, &checker); err != nil {
-		return nil, err
-	}
-
-	switch checker.IntegrationType() {
-	case APIGatewayRESTIntegration:
-		event := &events.APIGatewayProxyRequest{}
-		if err := json.Unmarshal(payload, event); err != nil {
-			return nil, err
-		}
-		if trace.RequestEvent != nil {
-			trace.RequestEvent(ctx, payload)
-		}
-		res, err = l.InvokeRESTAPI(ctx, event)
-	case APIGatewayHTTPIntegration:
-		event := &events.APIGatewayV2HTTPRequest{}
-		if err := json.Unmarshal(payload, event); err != nil {
-			return nil, err
-		}
-		if trace.RequestEvent != nil {
-			trace.RequestEvent(ctx, payload)
-		}
-		res, err = l.InvokeHTTPAPI(ctx, event)
-	case ALBTargetGroupIntegration:
-		event := &events.ALBTargetGroupRequest{}
-		if err := json.Unmarshal(payload, event); err != nil {
-			return nil, err
-		}
-		if trace.RequestEvent != nil {
-			trace.RequestEvent(ctx, payload)
-		}
-		res, err = l.InvokeALBTargetGroup(ctx, event)
-	case APIGatewayWebsocketIntegration:
-		event := &events.APIGatewayWebsocketProxyRequest{}
-		if err := json.Unmarshal(payload, event); err != nil {
-			return nil, err
+	if err = json.Unmarshal(payload, &checker); err != nil {
+		res, err = l.HandleNonHTTPEvent(ctx, payload, http.DetectContentType(payload))
+	} else {
+		switch checker.IntegrationType() {
+		case APIGatewayRESTIntegration:
+			event := &events.APIGatewayProxyRequest{}
+			if err := json.Unmarshal(payload, event); err != nil {
+				return nil, err
+			}
+			if trace.RequestEvent != nil {
+				trace.RequestEvent(ctx, payload)
+			}
+			res, err = l.InvokeRESTAPI(ctx, event)
+		case APIGatewayHTTPIntegration:
+			event := &events.APIGatewayV2HTTPRequest{}
+			if err := json.Unmarshal(payload, event); err != nil {
+				return nil, err
+			}
+			if trace.RequestEvent != nil {
+				trace.RequestEvent(ctx, payload)
+			}
+			res, err = l.InvokeHTTPAPI(ctx, event)
+		case ALBTargetGroupIntegration:
+			event := &events.ALBTargetGroupRequest{}
+			if err := json.Unmarshal(payload, event); err != nil {
+				return nil, err
+			}
+			if trace.RequestEvent != nil {
+				trace.RequestEvent(ctx, payload)
+			}
+			res, err = l.InvokeALBTargetGroup(ctx, event)
+		case APIGatewayWebsocketIntegration:
+			event := &events.APIGatewayWebsocketProxyRequest{}
+			if err := json.Unmarshal(payload, event); err != nil {
+				return nil, err
+			}
+			res, err = l.InvokeWebsocketAPI(ctx, event)
+		default:
+			res, err = l.HandleNonHTTPEvent(ctx, payload, "application/json")
 		}
-		res, err = l.InvokeWebsocketAPI(ctx, event)
-	default:
-		return nil, fmt.Errorf("unknown lambda integration type")
 	}
 
 	if err != nil {
@@ -208,9 +237,13 @@ func (l *LambdaHandler) Invoke(ctx context.Context, payload []byte) ([]byte, err
 		trace.ResponseEvent(ctx, res)
 	}
 
-	if responseBytes, err := json.Marshal(res); err != nil {
-		return nil, err
+	if b, ok := res.([]byte); ok {
+		return b, nil
 	} else {
-		return responseBytes, nil
+		if responseBytes, err := json.Marshal(res); err != nil {
+			return nil, err
+		} else {
+			return responseBytes, nil
+		}
 	}
 }

aws/lambda_passthrough.go

@@ -0,0 +1,45 @@
+/*
+Package aws provides an implementation using aws-sdk-go.
+
+Lambda event type compatibility layer for AWS API Gateway with HTTP API mode.
+
+See lambda event detail:
+https://docs.aws.amazon.com/apigateway/latest/developerguide/http-api-develop-integrations-lambda.html
+*/
+package aws
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"github.com/yacchi/lambda-http-adaptor/types"
+	"net/http"
+	"strconv"
+	"strings"
+)
+
+// NewLambdaPassthroughRequest Raw lambda event type to http.Request converter.
+func NewLambdaPassthroughRequest(ctx context.Context, payload []byte, eventPath string, contentType string) (r *http.Request, err error) {
+	var (
+		body   *bytes.Buffer
+		header = make(http.Header)
+	)
+
+	// build raw request URL
+	rawURL := "http://localhost/" + strings.TrimLeft(eventPath, "/")
+	body = bytes.NewBuffer(payload)
+
+	r, err = http.NewRequestWithContext(ctx, "POST", rawURL, body)
+	if err != nil {
+		return nil, fmt.Errorf("lambda_passthrough: new request: %w", err)
+	}
+
+	r.Header = header
+	r.Header.Set(types.HTTPHeaderContentType, contentType)
+	r.Header.Set(types.HTTPHeaderContentLength, strconv.Itoa(body.Len()))
+	r.RemoteAddr = "127.0.0.1"
+	r.RequestURI = r.URL.RequestURI()
+	r.Host = r.URL.Host
+
+	return
+}

example/simple/api/api.go

@@ -35,6 +35,17 @@ func (w *writerWrapper) WriteHeader(code int) {
 func ProvideAPI() http.Handler {
 	mux := http.NewServeMux()
 
+	mux.HandleFunc("/events", func(writer http.ResponseWriter, request *http.Request) {
+		writer.WriteHeader(http.StatusOK)
+		body, _ := io.ReadAll(request.Body)
+		println(request.Header.Get("Content-Type"))
+		println(string(body))
+		_, err := writer.Write(body)
+		if err != nil {
+			fmt.Println(err)
+		}
+	})
+
 	mux.HandleFunc("/websocket/$connect", func(writer http.ResponseWriter, request *http.Request) {
 		writer.WriteHeader(http.StatusOK)
 		err := json.NewEncoder(writer).Encode(map[string]interface{}{