Sender package added

Author: tzununbekov

go.mod

@@ -3,8 +3,10 @@ module github.com/triggermesh/aws-custom-runtime
 go 1.15
 
 require (
-	github.com/aws/aws-lambda-go v1.8.2
+	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.1.2
+	github.com/google/uuid v1.2.0
 	github.com/kelseyhightower/envconfig v1.4.0
+	knative.dev/eventing v0.22.1 // indirect
 )

go.sum

@@ -27,10 +27,7 @@ import (
 	"time"
 
 	"github.com/kelseyhightower/envconfig"
-	"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"
+	"github.com/triggermesh/aws-custom-runtime/pkg/sender"
 )
 
 var (
@@ -67,13 +64,15 @@ type Specification struct {
 	// Lambda API port to put function requests and get results
 	ExternalAPIport string `envconfig:"external_api_port" default:"8080"`
 
-	// 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"`
+	Sink           string `envconfig:"k_sink"`
+	ResponseFormat string `envconfig:"response_format"`
+}
+
+type Handler struct {
+	*sender.Handler
+
+	requestSizeLimit int64
+	functionTTL      int64
 }
 
 type message struct {
@@ -98,10 +97,10 @@ func (rw *responseWrapper) WriteHeader(statusCode int) {
 	rw.StatusCode = statusCode
 }
 
-func (s *Specification) setupEnv() error {
+func setupEnv(internalAPIport string) error {
 	environment["_HANDLER"], _ = os.LookupEnv("_HANDLER")
 	environment["LAMBDA_TASK_ROOT"], _ = os.LookupEnv("LAMBDA_TASK_ROOT")
-	environment["AWS_LAMBDA_RUNTIME_API"] += ":" + s.InternalAPIport
+	environment["AWS_LAMBDA_RUNTIME_API"] += ":" + internalAPIport
 
 	for k, v := range environment {
 		if err := os.Setenv(k, v); err != nil {
@@ -111,9 +110,9 @@ func (s *Specification) setupEnv() error {
 	return nil
 }
 
-func (s *Specification) newTask(w http.ResponseWriter, r *http.Request) {
-	requestSizeLimitInBytes := s.RequestSizeLimit * 1e+6
-	functionTTLInNanoSeconds := s.FunctionTTL * 1e+9
+func (h *Handler) newTask(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)
@@ -140,12 +139,19 @@ func (s *Specification) 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)))
+		// 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 {
+			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)
+		// w.WriteHeader(result.statusCode)
+		// w.Write(result.data)
+		if err := h.Send(result.data, result.statusCode, w); err != nil {
+			log.Printf("! %s %v\n", result.id, err)
+		}
 	}
 	mutex.Lock()
 	delete(results, task.id)
@@ -231,31 +237,6 @@ func responseHandler(w http.ResponseWriter, r *http.Request) {
 	return
 }
 
-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()
-		if err := envconfig.Process("CE", mapper); err != nil {
-			log.Fatalf("Cannot process CloudEvents wrapper env variables: %v", err)
-		}
-	default:
-		mapper = passthrough.NewMapper()
-	}
-
-	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		rw := responseWrapper{
-			ResponseWriter: w,
-		}
-		mapper.Request(r)
-		h.ServeHTTP(&rw, r)
-		mapper.Response(w, rw.StatusCode, rw.Body)
-	})
-}
-
 func ping(w http.ResponseWriter, r *http.Request) {
 	w.WriteHeader(http.StatusOK)
 	w.Write([]byte("pong"))
@@ -282,28 +263,39 @@ func api() error {
 }
 
 func main() {
+	// parse env
 	var spec Specification
 	if err := envconfig.Process("", &spec); err != nil {
 		log.Fatalf("Cannot process env variables: %v", err)
 	}
 	log.Printf("%+v\n", spec)
 
 	log.Println("Setting up runtime env")
-	if err := spec.setupEnv(); err != nil {
+	if err := setupEnv(spec.InternalAPIport); err != nil {
 		log.Fatalf("Cannot setup runime env: %v", err)
 	}
 
+	// create sender
+	sender := Handler{
+		Handler:          sender.New(spec.Sink, "text/plain"),
+		requestSizeLimit: spec.RequestSizeLimit,
+		functionTTL:      spec.FunctionTTL,
+	}
+
+	// setup channels
 	tasks = make(chan message, 100)
 	results = make(map[string]chan message)
 	defer close(tasks)
 
+	// start Lambda API
 	log.Println("Starting API")
 	go func() {
 		if err := api(); err != nil {
 			log.Fatalf("Runtime internal API error: %v", err)
 		}
 	}()
 
+	// start invokers
 	for i := 0; i < spec.NumberOfinvokers; i++ {
 		log.Println("Starting bootstrap", i+1)
 		go func(i int) {
@@ -317,9 +309,10 @@ func main() {
 		}(i)
 	}
 
+	// start external API
 	taskRouter := http.NewServeMux()
-	taskHandler := http.HandlerFunc(spec.newTask)
-	taskRouter.Handle("/", spec.mapEvent(taskHandler))
+	taskHandler := http.HandlerFunc(sender.newTask)
+	taskRouter.Handle("/", taskHandler)
 	log.Println("Listening...")
 	err := http.ListenAndServe(":"+spec.ExternalAPIport, taskRouter)
 	if err != nil && err != http.ErrServerClosed {

main.go

@@ -0,0 +1,46 @@
+package sender
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"net/http"
+)
+
+type Handler struct {
+	Target      string
+	ContentType string
+}
+
+func New(target, contentType string) *Handler {
+	return &Handler{
+		Target:      target,
+		ContentType: contentType,
+	}
+}
+
+func (h *Handler) Send(data []byte, statusCode int, writer http.ResponseWriter) error {
+	ctx := context.Background()
+
+	if h.Target != "" {
+		_, err := h.request(ctx, data)
+		if err != nil {
+			writer.WriteHeader(http.StatusInternalServerError)
+			return fmt.Errorf("failed to send the data: %w", err)
+		}
+		writer.WriteHeader(statusCode)
+		return nil
+	}
+
+	return h.reply(ctx, data, statusCode, writer)
+}
+
+func (h *Handler) request(ctx context.Context, data []byte) (*http.Response, error) {
+	return http.Post(h.Target, h.ContentType, bytes.NewBuffer(data))
+}
+
+func (h *Handler) reply(ctx context.Context, data []byte, statusCode int, writer http.ResponseWriter) error {
+	writer.WriteHeader(statusCode)
+	_, err := writer.Write(data)
+	return err
+}