Cloud events 0.1

cmd/sendevent/main.go

@@ -0,0 +1,97 @@
+/*
+Copyright 2018 The Knative Authors
+
+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
+
+    https://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.
+*/
+
+// Implements a simple utility for sending a JSON-encoded sample event.
+package main
+
+import (
+	"encoding/json"
+	"flag"
+	"fmt"
+	"io/ioutil"
+	"net/http"
+	"os"
+	"time"
+
+	"github.com/knative/eventing/pkg/event"
+
+	"github.com/google/uuid"
+)
+
+var (
+	context event.Context
+	webhook string
+	data    string
+)
+
+func init() {
+	flag.StringVar(&context.EventID, "event-id", "", "Event ID to use. Defaults to a generated UUID")
+	flag.StringVar(&context.EventType, "event-type", "google.events.action.demo", "The Event Type to use.")
+	flag.StringVar(&context.Source, "source", "", "Source URI to use. Defaults to the current machine's hostname")
+	flag.StringVar(&data, "data", `{"hello": "world!"}`, "Event data")
+}
+
+func main() {
+	flag.Parse()
+
+	if len(flag.Args()) != 1 {
+		fmt.Println("Usage: sendevent [flags] <webhook>\nFor details about valid flags, run sendevent --help")
+		os.Exit(1)
+	}
+
+	webhook := flag.Arg(0)
+
+	var untyped map[string]interface{}
+	if err := json.Unmarshal([]byte(data), &untyped); err != nil {
+		fmt.Println("Currently sendevent only supports JSON event data")
+		os.Exit(1)
+	}
+
+	fillEventContext(&context)
+	req, err := event.NewRequest(webhook, untyped, context)
+	if err != nil {
+		fmt.Printf("Failed to create request: %s", err)
+		os.Exit(1)
+	}
+
+	res, err := http.DefaultClient.Do(req)
+	if err != nil {
+		fmt.Printf("Failed to send event to %s: %s\n", webhook, err)
+		os.Exit(1)
+	}
+	fmt.Printf("Got response from %s\n%s\n", webhook, res.Status)
+	if res.Header.Get("Content-Length") != "" {
+		bytes, _ := ioutil.ReadAll(res.Body)
+		fmt.Println(string(bytes))
+	}
+}
+
+func fillEventContext(ctx *event.Context) {
+	ctx.CloudEventsVersion = "0.1"
+	ctx.EventTime = time.Now().UTC()
+
+	if ctx.EventID == "" {
+		ctx.EventID = uuid.New().String()
+	}
+
+	if ctx.Source == "" {
+		var err error
+		ctx.Source, err = os.Hostname()
+		if err != nil {
+			ctx.Source = "localhost"
+		}
+	}
+}

pkg/event/doc.go

@@ -0,0 +1,20 @@
+/*
+Copyright 2018 The Knative Authors
+
+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 event implements utilities for handling CloudEvents.
+// For information on the spec, see
+// https://github.com/cloudevents/spec/blob/v0.1/http-transport-binding.md
+package event

pkg/event/encoding_binary.go

@@ -0,0 +1,193 @@
+/*
+Copyright 2018 The Knative Authors
+
+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 event
+
+// TODO(inlined): must add header encoding/decoding
+
+import (
+	"bytes"
+	"encoding/json"
+	"fmt"
+	"io/ioutil"
+	"net/http"
+	"net/url"
+	"strings"
+
+	"github.com/golang/glog"
+)
+
+const (
+	// HeaderCloudEventsVersion is the header for the version of Cloud Events
+	// used.
+	HeaderCloudEventsVersion = "CE-CloudEventsVersion"
+
+	// HeaderEventID is the header for the unique ID of this event.
+	HeaderEventID = "CE-EventID"
+
+	// HeaderEventTime is the OPTIONAL header for the time at which an event
+	// occurred.
+	HeaderEventTime = "CE-EventTime"
+
+	// HeaderEventType is the header for type of event represented. Value SHOULD
+	// be in reverse-dns form.
+	HeaderEventType = "CE-EventType"
+
+	// HeaderEventTypeVersion is the OPTIONAL header for the version of the
+	// scheme for the event type.
+	HeaderEventTypeVersion = "CE-EventTypeVersion"
+
+	// HeaderSchemaURL is the OPTIONAL header for the schema of the event data.
+	HeaderSchemaURL = "CE-SchemaURL"
+
+	// HeaderSource is the header for the source which emitted this event.
+	HeaderSource = "CE-Source"
+
+	// HeaderExtensions is the OPTIONAL header prefix for CloudEvents extensions
+	headerExtensionsPrefix = "CE-X-"
+
+	// Binary implements Binary encoding/decoding
+	Binary binary = 0
+)
+
+type binary int
+
+// FromRequest parses event data and context from an HTTP request.
+func (binary) FromRequest(data interface{}, r *http.Request) (*Context, error) {
+	var ctx Context
+	err := anyError(
+		getRequiredHeader(r.Header, HeaderEventID, &ctx.EventID),
+		getRequiredHeader(r.Header, HeaderEventType, &ctx.EventType),
+		getRequiredHeader(r.Header, HeaderSource, &ctx.Source),
+		getRequiredHeader(r.Header, HeaderContentType, &ctx.ContentType))
+	if err != nil {
+		return nil, err
+	}
+
+	ctx.CloudEventsVersion = r.Header.Get(HeaderCloudEventsVersion)
+	if timeStr := r.Header.Get(HeaderEventTime); timeStr != "" {
+		if err := ctx.EventTime.UnmarshalText([]byte(timeStr)); err != nil {
+			return nil, err
+		}
+	}
+	ctx.EventTypeVersion = r.Header.Get(HeaderEventTypeVersion)
+	ctx.SchemaURL = r.Header.Get(HeaderSchemaURL)
+	if ctx.CloudEventsVersion != CloudEventsVersion {
+		glog.Warningf("Received CloudEvent version %q; parsing as version %q",
+			ctx.CloudEventsVersion, CloudEventsVersion)
+	}
+
+	ctx.Extensions = make(map[string]interface{})
+	for k, v := range r.Header {
+		if strings.ToUpper(k)[:len(headerExtensionsPrefix)] != headerExtensionsPrefix {
+			continue
+		}
+		name := k[len(headerExtensionsPrefix):]
+		var val interface{}
+		if err := json.Unmarshal([]byte(v[0]), &val); err != nil {
+			// If this is not a JSON object, treat it as a string.
+			// It's not clear when we would treat this as Bytes.
+			ctx.Extensions[name] = v[0]
+		} else {
+			ctx.Extensions[name] = val
+		}
+	}
+
+	if err := unmarshalEventData(ctx.ContentType, r.Body, data); err != nil {
+		return nil, err
+	}
+
+	return &ctx, nil
+}
+
+// NewRequest creates an HTTP request for Binary content encoding.
+func (binary) NewRequest(urlString string, data interface{}, context Context) (*http.Request, error) {
+	url, err := url.Parse(urlString)
+	if err != nil {
+		return nil, err
+	}
+
+	if err := ensureRequiredFields(context); err != nil {
+		return nil, err
+	}
+	// Defaultable values:
+	ceVersion := context.CloudEventsVersion
+	if ceVersion == "" {
+		ceVersion = CloudEventsVersion
+	}
+	contentType := context.ContentType
+	if contentType == "" {
+		contentType = contentTypeJSON
+	}
+
+	// non-string values:
+	eventTime := ""
+	if !context.EventTime.IsZero() {
+		b, err := context.EventTime.UTC().MarshalText()
+		if err != nil {
+			return nil, err
+		}
+		eventTime = string(b)
+	}
+
+	h := http.Header{}
+	setHeader(h, HeaderCloudEventsVersion, ceVersion)
+	setHeader(h, HeaderEventID, context.EventID)
+	setHeader(h, HeaderEventTime, eventTime)
+	setHeader(h, HeaderEventType, context.EventType)
+	setHeader(h, HeaderEventTypeVersion, context.EventTypeVersion)
+	setHeader(h, HeaderSchemaURL, context.SchemaURL)
+	setHeader(h, HeaderContentType, contentType)
+	setHeader(h, HeaderSource, context.Source)
+	for name, value := range context.Extensions {
+		encoded, err := json.Marshal(value)
+		if err != nil {
+			return nil, err
+		}
+		h[headerExtensionsPrefix+name] = []string{
+			string(encoded),
+		}
+	}
+
+	b, err := marshalEventData(contentType, data)
+	if err != nil {
+		return nil, err
+	}
+
+	return &http.Request{
+		Method: http.MethodPost,
+		URL:    url,
+		Header: h,
+		Body:   ioutil.NopCloser(bytes.NewReader(b)),
+	}, nil
+}
+
+// TODO(inlined) URI encoding/decoding of headers
+func getHeader(h http.Header, name string) string {
+	return h.Get(name)
+}
+
+func setHeader(h http.Header, name string, value string) {
+	if value != "" {
+		h.Set(name, value)
+	}
+}
+func getRequiredHeader(h http.Header, name string, value *string) error {
+	if *value = getHeader(h, name); *value == "" {
+		return fmt.Errorf("missing required header %q", name)
+	}
+	return nil
+}