Converter interface added

Author: tzununbekov

Makefile

@@ -18,7 +18,7 @@ GOLINT            ?= golangci-lint run
 GOTOOL            ?= go tool
 GOTEST            ?= gotestsum --junitfile $(TEST_OUTPUT_DIR)/$(PACKAGE)-unit-tests.xml --format pkgname-and-test-fails --
 
-GOPKGS             = ./pkg/events/...
+GOPKGS             = ./pkg/...
 LDFLAGS            = -extldflags=-static -w -s
 
 HAS_GOTESTSUM     := $(shell command -v gotestsum;)

go.mod

@@ -3,10 +3,6 @@ module github.com/triggermesh/aws-custom-runtime
 go 1.15
 
 require (
-	github.com/aws/aws-lambda-go v1.13.3
-	github.com/cloudevents/sdk-go v1.2.0 // indirect
-	github.com/cloudevents/sdk-go/v2 v2.3.1
 	github.com/google/uuid v1.2.0
 	github.com/kelseyhightower/envconfig v1.4.0
-	knative.dev/eventing v0.22.1 // indirect
 )

go.sum

@@ -1,16 +1,18 @@
-// Copyright 2018 TriggerMesh, Inc
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
+/*
+Copyright 2021 Triggermesh Inc.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
 
 package main
 
@@ -27,6 +29,8 @@ import (
 	"time"
 
 	"github.com/kelseyhightower/envconfig"
+
+	"github.com/triggermesh/aws-custom-runtime/pkg/converter"
 	"github.com/triggermesh/aws-custom-runtime/pkg/sender"
 )
 
@@ -69,7 +73,8 @@ type Specification struct {
 }
 
 type Handler struct {
-	*sender.Handler
+	Sender    *sender.Sender
+	Converter converter.Converter
 
 	requestSizeLimit int64
 	functionTTL      int64
@@ -139,17 +144,17 @@ func (h *Handler) newTask(w http.ResponseWriter, r *http.Request) {
 	select {
 	case <-time.After(time.Duration(functionTTLInNanoSeconds)):
 		log.Printf("-> ! %s Deadline is reached\n", task.id)
-		// w.WriteHeader(http.StatusGone)
-		// w.Write([]byte(fmt.Sprintf("Deadline is reached, data %s", task.data)))
 		resp := []byte(fmt.Sprintf("Deadline is reached, data %s", task.data))
-		if err := h.Send(resp, http.StatusGone, w); err != nil {
+		if err := h.Sender.Send(resp, http.StatusGone, w); err != nil {
 			log.Printf("! %s %v\n", task.id, err)
 		}
 	case result := <-resultsChannel:
 		log.Printf("-> %s %d %s\n", result.id, result.statusCode, result.data)
-		// w.WriteHeader(result.statusCode)
-		// w.Write(result.data)
-		if err := h.Send(result.data, result.statusCode, w); err != nil {
+		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)
 		}
 	}
@@ -275,9 +280,16 @@ func main() {
 		log.Fatalf("Cannot setup runime env: %v", err)
 	}
 
-	// create sender
-	sender := Handler{
-		Handler:          sender.New(spec.Sink, "text/plain"),
+	// create converter
+	conv, err := converter.New(spec.ResponseFormat)
+	if err != nil {
+		log.Fatalf("Cannot create converter: %v", err)
+	}
+
+	// setup sender
+	handler := Handler{
+		Sender:           sender.New(spec.Sink, conv.ContentType()),
+		Converter:        conv,
 		requestSizeLimit: spec.RequestSizeLimit,
 		functionTTL:      spec.FunctionTTL,
 	}
@@ -311,10 +323,10 @@ func main() {
 
 	// start external API
 	taskRouter := http.NewServeMux()
-	taskHandler := http.HandlerFunc(sender.newTask)
+	taskHandler := http.HandlerFunc(handler.newTask)
 	taskRouter.Handle("/", 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)
 	}

main.go

@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"fmt"
 	"io/ioutil"
+	"log"
 	"net/http"
 	"net/http/httptest"
 	"os"
@@ -14,6 +15,8 @@ import (
 	"time"
 
 	"github.com/kelseyhightower/envconfig"
+	"github.com/triggermesh/aws-custom-runtime/pkg/converter"
+	"github.com/triggermesh/aws-custom-runtime/pkg/sender"
 )
 
 func TestSetupEnv(t *testing.T) {
@@ -23,7 +26,7 @@ func TestSetupEnv(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	if err := s.setupEnv(); err != nil {
+	if err := setupEnv(s.InternalAPIport); err != nil {
 		t.Errorf("Setup Env error %s\n", err)
 	}
 
@@ -41,21 +44,33 @@ func TestNewTask(t *testing.T) {
 		t.Fatal(err)
 	}
 
+	conv, err := converter.New(s.ResponseFormat)
+	if err != nil {
+		log.Fatalf("Cannot create converter: %v", err)
+	}
+
+	handler := Handler{
+		Sender:           sender.New(s.Sink, conv.ContentType()),
+		Converter:        conv,
+		requestSizeLimit: s.RequestSizeLimit,
+		functionTTL:      s.FunctionTTL,
+	}
+
 	payload := []byte(`{"payload": "test"}`)
 
 	tasks = make(chan message, 100)
 	results = make(map[string]chan message)
 	defer close(tasks)
 
 	recorder := httptest.NewRecorder()
-	handler := http.HandlerFunc(s.newTask)
+	h := http.HandlerFunc(handler.newTask)
 
 	req, err := http.NewRequest("POST", "/", bytes.NewBuffer(payload))
 	if err != nil {
 		t.Fatal(err)
 	}
 
-	go handler.ServeHTTP(recorder, req)
+	go h.ServeHTTP(recorder, req)
 	task := <-tasks
 	results[task.id] <- task
 	time.Sleep(time.Millisecond * 100)

main_test.go

@@ -0,0 +1,20 @@
+/*
+Copyright 2021 Triggermesh Inc.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package apigateway
+
+// TODO: move logic from
+// https://github.com/triggermesh/aws-custom-runtime/blob/b3ec30de10c29b761fc1d89e9dd828d554c1ce0f/pkg/events/apigateway/mapper.go

pkg/converter/apigateway/apigateway.go

@@ -0,0 +1,68 @@
+/*
+Copyright 2021 Triggermesh Inc.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package cloudevents
+
+import (
+	"encoding/json"
+	"fmt"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/kelseyhightower/envconfig"
+)
+
+type ceBody struct {
+	ID          string `json:"id"`
+	Type        string `json:"type"`
+	Time        string `json:"time"`
+	Source      string `json:"source"`
+	Specversion string `json:"specversion"`
+	Data        string `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"`
+	Source    string `envconfig:"source" default:"knative-lambda-runtime"`
+	Subject   string `envconfig:"subject" default:"klr-response"`
+}
+
+func New() (*CloudEvent, error) {
+	var ce CloudEvent
+	if err := envconfig.Process("ce", &ce); err != nil {
+		return nil, fmt.Errorf("Cannot process CloudEvent env variables: %v", err)
+	}
+	return &ce, nil
+}
+
+func (ce *CloudEvent) Convert(data []byte) ([]byte, error) {
+	b := ceBody{
+		ID:          uuid.NewString(),
+		Type:        ce.EventType,
+		Time:        time.Now().Format(time.RFC3339),
+		Source:      ce.Source,
+		Specversion: "1.0",
+		Data:        string(data),
+	}
+	return json.Marshal(b)
+}
+
+func (ce *CloudEvent) ContentType() string {
+	return contentType
+}

pkg/converter/cloudevents/cloudevents.go

@@ -0,0 +1,36 @@
+/*
+Copyright 2021 Triggermesh Inc.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package converter
+
+import (
+	"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)
+	ContentType() string
+}
+
+func New(format string) (Converter, error) {
+	switch format {
+	case "API_GATEWAY":
+	case "CLOUDEVENTS":
+		return cloudevents.New()
+	}
+	return plain.New()
+}

pkg/converter/converter.go

@@ -0,0 +1,33 @@
+/*
+Copyright 2021 Triggermesh Inc.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package plain
+
+type Plain struct{}
+
+const contentType = "plain/text"
+
+func New() (*Plain, error) {
+	return &Plain{}, nil
+}
+
+func (p *Plain) Convert(data []byte) ([]byte, error) {
+	return data, nil
+}
+
+func (p *Plain) ContentType() string {
+	return contentType
+}