Merge pull request #32 from triggermesh/request-conversion

Request conversion

Author: tzununbekov

main.go

@@ -84,24 +84,10 @@ type message struct {
 	id         string
 	deadline   int64
 	data       []byte
+	context    map[string]string
 	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
-}
-
-func (rw *responseWrapper) WriteHeader(statusCode int) {
-	rw.StatusCode = statusCode
-}
-
 func setupEnv(internalAPIport string) error {
 	environment["_HANDLER"], _ = os.LookupEnv("_HANDLER")
 	environment["LAMBDA_TASK_ROOT"], _ = os.LookupEnv("LAMBDA_TASK_ROOT")
@@ -115,21 +101,38 @@ func setupEnv(internalAPIport string) error {
 	return nil
 }
 
-func (h *Handler) newTask(w http.ResponseWriter, r *http.Request) {
+func (h *Handler) serve(w http.ResponseWriter, r *http.Request) {
 	requestSizeLimitInBytes := h.requestSizeLimit * 1e+6
-	functionTTLInNanoSeconds := h.functionTTL * 1e+9
 	body, err := ioutil.ReadAll(http.MaxBytesReader(w, r.Body, requestSizeLimitInBytes))
 	if err != nil {
-		http.Error(w, err.Error(), http.StatusInternalServerError)
+		http.Error(w, err.Error(), http.StatusRequestEntityTooLarge)
 		return
 	}
 	defer r.Body.Close()
 
+	req, context, err := h.Converter.Request(body, r.Header)
+	if err != nil {
+		http.Error(w, err.Error(), http.StatusInternalServerError)
+		return
+	}
+
+	result := enqueue(req, context, h.functionTTL*1e+9)
+	result.data, err = h.Converter.Response(result.data)
+	if err != nil {
+		result.data = []byte(fmt.Sprintf("Response conversion error: %v", err))
+	}
+	if err := h.Sender.Send(result.data, result.statusCode, w); err != nil {
+		log.Printf("! %s %s %v\n", result.id, result.data, err)
+	}
+}
+
+func enqueue(request []byte, context map[string]string, ttl int64) message {
 	now := time.Now().UnixNano()
 	task := message{
 		id:       fmt.Sprintf("%d", now),
-		deadline: now + functionTTLInNanoSeconds,
-		data:     body,
+		deadline: now + ttl,
+		data:     request,
+		context:  context,
 	}
 	log.Printf("<- %s %s\n", task.id, task.data)
 
@@ -141,27 +144,22 @@ func (h *Handler) newTask(w http.ResponseWriter, r *http.Request) {
 
 	tasks <- task
 
+	var resp message
 	select {
-	case <-time.After(time.Duration(functionTTLInNanoSeconds)):
-		log.Printf("-> ! %s Deadline is reached\n", task.id)
-		resp := []byte(fmt.Sprintf("Deadline is reached, data %s", task.data))
-		if err := h.Sender.Send(resp, http.StatusGone, w); err != nil {
-			log.Printf("! %s %v\n", task.id, err)
+	case <-time.After(time.Duration(ttl)):
+		resp = message{
+			id:         task.id,
+			data:       []byte(fmt.Sprintf("Deadline is reached, data %s", task.data)),
+			statusCode: http.StatusGone,
 		}
 	case result := <-resultsChannel:
-		log.Printf("-> %s %d %s\n", result.id, result.statusCode, result.data)
-		body, err := h.Converter.Convert(result.data)
-		if err != nil {
-			log.Printf("! %s %v\n", result.id, err)
-		}
-		if err := h.Sender.Send(body, result.statusCode, w); err != nil {
-			log.Printf("! %s %v\n", result.id, err)
-		}
+		resp = result
 	}
 	mutex.Lock()
 	delete(results, task.id)
 	mutex.Unlock()
-	return
+	log.Printf("-> %s %d %s\n", resp.id, resp.statusCode, resp.data)
+	return resp
 }
 
 func getTask(w http.ResponseWriter, r *http.Request) {
@@ -172,6 +170,9 @@ func getTask(w http.ResponseWriter, r *http.Request) {
 	w.Header().Set("Lambda-Runtime-Deadline-Ms", strconv.Itoa(int(task.deadline)))
 	w.Header().Set("Lambda-Runtime-Invoked-Function-Arn", "arn:aws:lambda:us-east-1:123456789012:function:custom-runtime")
 	w.Header().Set("Lambda-Runtime-Trace-Id", "0")
+	for k, v := range task.context {
+		w.Header().Set(k, v)
+	}
 
 	w.WriteHeader(http.StatusOK)
 	w.Write(task.data)
@@ -323,8 +324,7 @@ func main() {
 
 	// start external API
 	taskRouter := http.NewServeMux()
-	taskHandler := http.HandlerFunc(handler.newTask)
-	taskRouter.Handle("/", taskHandler)
+	taskRouter.Handle("/", http.HandlerFunc(handler.serve))
 	log.Println("Listening...")
 	err = http.ListenAndServe(":"+spec.ExternalAPIport, taskRouter)
 	if err != nil && err != http.ErrServerClosed {

main_test.go

@@ -63,7 +63,7 @@ func TestNewTask(t *testing.T) {
 	defer close(tasks)
 
 	recorder := httptest.NewRecorder()
-	h := http.HandlerFunc(handler.newTask)
+	h := http.HandlerFunc(handler.serve)
 
 	req, err := http.NewRequest("POST", "/", bytes.NewBuffer(payload))
 	if err != nil {

pkg/converter/cloudevents/cloudevents.go

@@ -17,25 +17,29 @@ limitations under the License.
 package cloudevents
 
 import (
+	"bytes"
 	"encoding/json"
 	"fmt"
+	"net/http"
+	"strings"
 	"time"
 
 	"github.com/google/uuid"
 	"github.com/kelseyhightower/envconfig"
 )
 
+const contentType = "application/cloudevents+json"
+
 type ceBody struct {
 	ID          string      `json:"id"`
 	Type        string      `json:"type"`
 	Time        string      `json:"time"`
 	Source      string      `json:"source"`
 	Specversion string      `json:"specversion"`
+	Contenttype string      `json:"datacontenttype"`
 	Data        interface{} `json:"data"`
 }
 
-const contentType = "application/cloudevents+json"
-
 // CloudEvent is a data structure required to map KLR responses to cloudevents
 type CloudEvent struct {
 	EventType string `envconfig:"type" default:"ce.klr.triggermesh.io"`
@@ -51,7 +55,7 @@ func New() (*CloudEvent, error) {
 	return &ce, nil
 }
 
-func (ce *CloudEvent) Convert(data []byte) ([]byte, error) {
+func (ce *CloudEvent) Response(data []byte) ([]byte, error) {
 	// If response format is set to CloudEvents
 	// and CE_TYPE is empty,
 	// then reply with the empty response
@@ -60,11 +64,18 @@ func (ce *CloudEvent) Convert(data []byte) ([]byte, error) {
 	}
 
 	var body interface{}
-	body = string(data)
+	contentType := "text/plain"
 
-	// try to decode function's response into JSON
-	if json.Valid(data) {
+	switch {
+	case json.Valid(data) &&
+		(bytes.TrimSpace(data)[0] == '{' ||
+			bytes.TrimSpace(data)[0] == '['):
+		contentType = "application/json"
 		body = json.RawMessage(data)
+	default:
+		data = bytes.TrimSpace(data)
+		data = bytes.Trim(data, "\"")
+		body = string(data)
 	}
 
 	b := ceBody{
@@ -73,11 +84,71 @@ func (ce *CloudEvent) Convert(data []byte) ([]byte, error) {
 		Time:        time.Now().Format(time.RFC3339),
 		Source:      ce.Source,
 		Specversion: "1.0",
+		Contenttype: contentType,
 		Data:        body,
 	}
 	return json.Marshal(b)
 }
 
+func (ce *CloudEvent) Request(request []byte, headers http.Header) ([]byte, map[string]string, error) {
+	var context map[string]string
+	var body []byte
+	var err error
+
+	switch headers.Get("Content-Type") {
+	case "application/cloudevents+json":
+		if body, context, err = parseStructuredCE(request); err != nil {
+			return nil, nil, fmt.Errorf("structured CloudEvent parse error: %w", err)
+		}
+	case "application/json":
+		body = request
+		context = parseBinaryCE(headers)
+	default:
+		return request, nil, nil
+	}
+
+	ceContext, err := json.Marshal(context)
+	if err != nil {
+		return nil, nil, fmt.Errorf("cannot encode request context: %w", err)
+	}
+
+	runtimeContext := map[string]string{
+		"Lambda-Runtime-Cloudevents-Context": string(ceContext),
+	}
+
+	return body, runtimeContext, nil
+}
+
+func parseStructuredCE(body []byte) ([]byte, map[string]string, error) {
+	var event map[string]interface{}
+	if err := json.Unmarshal(body, &event); err != nil {
+		return nil, nil, fmt.Errorf("cannot unmarshal body: %w", err)
+	}
+
+	data, err := json.Marshal(event["data"])
+	if err != nil {
+		return nil, nil, fmt.Errorf("cannot marshal body: %w", err)
+	}
+
+	delete(event, "data")
+	headers := make(map[string]string, len(event))
+	for k, v := range event {
+		headers[k] = fmt.Sprintf("%v", v)
+	}
+
+	return data, headers, nil
+}
+
+func parseBinaryCE(headers http.Header) map[string]string {
+	h := make(map[string]string)
+	for k, v := range headers {
+		if strings.HasPrefix(k, "Ce-") {
+			h[strings.ToLower(k[3:])] = strings.Join(v, ",")
+		}
+	}
+	return h
+}
+
 func (ce *CloudEvent) ContentType() string {
 	return contentType
 }

pkg/converter/converter.go

@@ -17,12 +17,15 @@ limitations under the License.
 package converter
 
 import (
+	"net/http"
+
 	"github.com/triggermesh/aws-custom-runtime/pkg/converter/cloudevents"
 	"github.com/triggermesh/aws-custom-runtime/pkg/converter/plain"
 )
 
 type Converter interface {
-	Convert([]byte) ([]byte, error)
+	Response([]byte) ([]byte, error)
+	Request([]byte, http.Header) ([]byte, map[string]string, error)
 	ContentType() string
 }
 

pkg/converter/plain/cloudevents.go renamed to pkg/converter/plain/plain.go

@@ -16,6 +16,8 @@ limitations under the License.
 
 package plain
 
+import "net/http"
+
 type Plain struct{}
 
 const contentType = "plain/text"
@@ -24,10 +26,14 @@ func New() (*Plain, error) {
 	return &Plain{}, nil
 }
 
-func (p *Plain) Convert(data []byte) ([]byte, error) {
+func (p *Plain) Response(data []byte) ([]byte, error) {
 	return data, nil
 }
 
+func (p *Plain) Request(request []byte, headers http.Header) ([]byte, map[string]string, error) {
+	return request, nil, nil
+}
+
 func (p *Plain) ContentType() string {
 	return contentType
 }