Mapper interface added

Author: tzununbekov

main.go

@@ -27,43 +27,13 @@ import (
 	"time"
 
 	"github.com/kelseyhightower/envconfig"
-	"github.com/triggermesh/aws-custom-runtime/pkg/events/apiGateway"
+	"github.com/triggermesh/aws-custom-runtime/pkg/events"
+	"github.com/triggermesh/aws-custom-runtime/pkg/events/apigateway"
+	"github.com/triggermesh/aws-custom-runtime/pkg/events/cloudevents"
+	"github.com/triggermesh/aws-custom-runtime/pkg/events/passthrough"
 )
 
-type message struct {
-	id         string
-	deadline   int64
-	data       []byte
-	statusCode int
-}
-
-type responseWrapper struct {
-	http.ResponseWriter
-	StatusCode int
-	Body       []byte
-}
-
-// Specification is a set of env variables that can be used to configure runtime API
-type Specification struct {
-	// Number of bootstrap processes
-	NumberOfinvokers int `envconfig:"invoker_count" default:"4"`
-	// Request body size limit, Mb
-	RequestSizeLimit int64 `envconfig:"request_size_limit" default:"5"`
-	// Funtions deadline, seconds
-	FunctionTTL int64 `envconfig:"function_ttl" default:"10"`
-	// Lambda runtime API port for functions
-	InternalAPIport string `envconfig:"internal_api_port" default:"80"`
-	// Lambda API port to put function requests and get results
-	ExternalAPIport string `envconfig:"external_api_port" default:"8080"`
-	// Either return function result "as is" or consider it as API Gateway JSON
-	EventType string `envconfig:"event_type"`
-}
-
 var (
-	// numberOfinvokers = 4
-	// requestSizeLimit int64 = 5
-	// functionTTL      int64 = 10
-
 	tasks   chan message
 	results map[string]chan message
 
@@ -84,6 +54,43 @@ var (
 	}
 )
 
+// Specification is a set of env variables that can be used to configure runtime API
+type Specification struct {
+	// Number of bootstrap processes
+	NumberOfinvokers int `envconfig:"invoker_count" default:"4"`
+	// Request body size limit, Mb
+	RequestSizeLimit int64 `envconfig:"request_size_limit" default:"5"`
+	// Funtions deadline, seconds
+	FunctionTTL int64 `envconfig:"function_ttl" default:"10"`
+	// Lambda runtime API port for functions
+	InternalAPIport string `envconfig:"internal_api_port" default:"80"`
+	// Lambda API port to put function requests and get results
+	ExternalAPIport string `envconfig:"external_api_port" default:"8080"`
+	// Parent Knative Service name
+	Service string `envconfig:"k_service"`
+
+	// Apply response wrapping before sending it back to the client.
+	// Common case - AWS Lambda functions usually returns data formatted for API Gateway service.
+	// Set "RESPONSE_WRAPPER: API_GATEWAY" and receive events as if they were processed by API Gateway.
+	// Opposite scenario - return responses in CloudEvent format: "RESPONSE_WRAPPER: CLOUDEVENTS"
+	// NOTE: Response wrapper does both encoding and decoding depending on the type. We should consider
+	// separating wrappers by their function.
+	ResponseWrapper string `envconfig:"response_wrapper"`
+}
+
+type message struct {
+	id         string
+	deadline   int64
+	data       []byte
+	statusCode int
+}
+
+type responseWrapper struct {
+	http.ResponseWriter
+	StatusCode int
+	Body       []byte
+}
+
 func (rw *responseWrapper) Write(data []byte) (int, error) {
 	rw.Body = data
 	return len(data), nil
@@ -227,16 +234,24 @@ func responseHandler(w http.ResponseWriter, r *http.Request) {
 }
 
 func (s *Specification) mapEvent(h http.Handler) http.Handler {
+	var mapper events.Mapper
+
+	switch s.ResponseWrapper {
+	case "API_GATEWAY":
+		mapper = apigateway.NewMapper()
+	case "CLOUDEVENTS":
+		mapper = cloudevents.NewMapper(s.Service)
+	default:
+		mapper = passthrough.NewMapper()
+	}
+
 	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		rw := responseWrapper{w, 200, []byte{}}
-		switch s.EventType {
-		case "API_GATEWAY":
-			apiGateway.Request(r)
-			h.ServeHTTP(&rw, r)
-			apiGateway.Response(w, rw.StatusCode, rw.Body)
-		default:
-			h.ServeHTTP(w, r)
+		rw := responseWrapper{
+			ResponseWriter: w,
 		}
+		mapper.Request(r)
+		h.ServeHTTP(&rw, r)
+		mapper.Response(w, rw.StatusCode, rw.Body)
 	})
 }
 
@@ -266,23 +281,21 @@ func api() error {
 }
 
 func main() {
-	tasks = make(chan message, 100)
-	results = make(map[string]chan message)
-	defer close(tasks)
-
 	var spec Specification
-
-	err := envconfig.Process("", &spec)
-	if err != nil {
+	if err := envconfig.Process("", &spec); err != nil {
 		log.Fatalf("Cannot process env variables: %v", err)
 	}
 	log.Printf("%+v\n", spec)
 
-	log.Println("Setup app env")
+	log.Println("Setting up runtime env")
 	if err := spec.setupEnv(); err != nil {
 		log.Fatalf("Cannot setup runime env: %v", err)
 	}
 
+	tasks = make(chan message, 100)
+	results = make(map[string]chan message)
+	defer close(tasks)
+
 	log.Println("Starting API")
 	go func() {
 		if err := api(); err != nil {
@@ -307,7 +320,7 @@ func main() {
 	taskHandler := http.HandlerFunc(spec.newTask)
 	taskRouter.Handle("/", spec.mapEvent(taskHandler))
 	log.Println("Listening...")
-	err = http.ListenAndServe(":"+spec.ExternalAPIport, taskRouter)
+	err := http.ListenAndServe(":"+spec.ExternalAPIport, taskRouter)
 	if err != nil && err != http.ErrServerClosed {
 		log.Fatalf("Runtime external API error: %v", err)
 	}

pkg/events/apiGateway/mapper.go renamed to pkg/events/apigateway/mapper.go

@@ -1,4 +1,4 @@
-package apiGateway
+package apigateway
 
 import (
 	"bytes"
@@ -11,7 +11,13 @@ import (
 	"github.com/aws/aws-lambda-go/events"
 )
 
-func Request(r *http.Request) {
+type APIGateway struct{}
+
+func NewMapper() *APIGateway {
+	return &APIGateway{}
+}
+
+func (a *APIGateway) Request(r *http.Request) {
 	event := events.APIGatewayProxyRequest{}
 	body, err := ioutil.ReadAll(r.Body)
 	if err != nil {
@@ -40,7 +46,7 @@ func Request(r *http.Request) {
 	r.Body = ioutil.NopCloser(bytes.NewBuffer(js))
 }
 
-func Response(w http.ResponseWriter, statusCode int, data []byte) (int, error) {
+func (a *APIGateway) Response(w http.ResponseWriter, statusCode int, data []byte) (int, error) {
 	var js events.APIGatewayProxyResponse
 	if err := json.Unmarshal(data, &js); err != nil {
 		return 0, err

pkg/events/cloudevents/mapper.go

@@ -0,0 +1,32 @@
+package cloudevents
+
+import (
+	"fmt"
+	"net/http"
+
+	cloudevents "github.com/cloudevents/sdk-go/v2"
+	"github.com/google/uuid"
+)
+
+type CloudEvents struct {
+	Service string
+}
+
+func NewMapper(service string) *CloudEvents {
+	return &CloudEvents{
+		Service: service,
+	}
+}
+
+func (c *CloudEvents) Request(r *http.Request) {}
+
+func (c *CloudEvents) Response(w http.ResponseWriter, statusCode int, data []byte) (int, error) {
+	response := cloudevents.NewEvent()
+	response.SetType("ce.klr.triggermesh.io")
+	response.SetSource(c.Service)
+	response.SetID(uuid.New().String())
+	if err := response.SetData(cloudevents.ApplicationJSON, data); err != nil {
+		return 0, fmt.Errorf("failed to set event data: %w", err)
+	}
+	return w.Write([]byte(response.String()))
+}

pkg/events/interface.go

@@ -0,0 +1,8 @@
+package events
+
+import "net/http"
+
+type Mapper interface {
+	Request(r *http.Request)
+	Response(w http.ResponseWriter, statusCode int, data []byte) (int, error)
+}

pkg/events/passthrough/mapper.go

@@ -0,0 +1,18 @@
+package passthrough
+
+import (
+	"net/http"
+)
+
+type Passthrough struct{}
+
+func NewMapper() *Passthrough {
+	return &Passthrough{}
+}
+
+func (c *Passthrough) Request(r *http.Request) {}
+
+func (c *Passthrough) Response(w http.ResponseWriter, statusCode int, data []byte) (int, error) {
+	w.WriteHeader(statusCode)
+	return w.Write(data)
+}