Serverless連載5: CloudEventsのGo版SDKをいじってみる

package main

import (
	"context"
	"fmt"
	"log"

	cloudevents "github.com/cloudevents/sdk-go"
)

func Receive(event cloudevents.Event) {
	// do something with event.Context and event.Data (via event.DataAs(foo)
	fmt.Printf("Event received.\n====\n%s====\n", event)
}

func main() {
	c, err := cloudevents.NewDefaultClient()
	if err != nil {
		log.Fatalf("failed to create client, %v", err)
	}
	log.Fatal(c.StartReceiver(context.Background(), Receive))
}

package main

import (
	"context"

	cloudevents "github.com/cloudevents/sdk-go"
)

func main() {
	event := cloudevents.NewEvent()
	event.SetID("ABC-123")
	event.SetType("com.cloudevents.readme.sent")
	event.SetSource("somesource")
	event.SetData(map[string]string{"hello": "world"})

	t, err := cloudevents.NewHTTPTransport(
		cloudevents.WithTarget("http://localhost:8080/"),
		cloudevents.WithEncoding(cloudevents.HTTPBinaryV1),
	)
	if err != nil {
		log.Fatal("failed to create transport, " + err.Error())
	}

	c, err := cloudevents.NewClient(t)
	if err != nil {
		log.Fatal("unable to create cloudevent client: " + err.Error())
	}

	_, _, err = c.Send(context.Background(), event)
	if err != nil {
		log.Fatal("failed to send cloudevent: " + err.Error())
	}
}

package main

import (
	"context"
	"fmt"
	"log"
	"os"

	cloudevents "github.com/cloudevents/sdk-go"
	"github.com/cloudevents/sdk-go/pkg/cloudevents/client"
	cepubsub "github.com/cloudevents/sdk-go/pkg/cloudevents/transport/pubsub"
	pscontext "github.com/cloudevents/sdk-go/pkg/cloudevents/transport/pubsub/context"
	"github.com/kelseyhightower/envconfig"
)

type envConfig struct {
	ProjectID string `envconfig:"GOOGLE_CLOUD_PROJECT"`

	TopicID string `envconfig:"PUBSUB_TOPIC" default:"demo_cloudevents" required:"true"`

	SubscriptionID string `envconfig:"PUBSUB_SUBSCRIPTION" default:"demo_cloudevents_subscriber" required:"true"`
}

type Model struct {
	Sequence int    `json:"id"`
	Message  string `json:"message"`
}

func receive(ctx context.Context, event cloudevents.Event, resp *cloudevents.EventResponse) error {
	fmt.Printf("Event Context: %+v\n", event.Context)
	fmt.Printf("Transport Context: %+v\n", pscontext.TransportContextFrom(ctx))

	data := &Model{}
	if err := event.DataAs(data); err != nil {
		fmt.Printf("Got Data Error: %s\n", err.Error())
	}
	fmt.Printf("Data: %+v\n", data)

	fmt.Printf("----------------------------\n")
	return nil
}

func main() {
	ctx := context.Background()

	var env envConfig
	if err := envconfig.Process("", &env); err != nil {
		log.Fatalf("[ERROR] Failed to process env var: %s", err)
	}

	log.Printf("[INFO] ProjectID: %s", env.ProjectID)
	log.Printf("[INFO] TopicID: %s", env.TopicID)
	log.Printf("[INFO] SubscriptionID: %s", env.SubscriptionID)
	t, err := cepubsub.New(context.Background(),
		cepubsub.WithProjectID(env.ProjectID),
		cepubsub.WithSubscriptionID(env.SubscriptionID))
	if err != nil {
		log.Fatalf("failed to create pubsub transport, %s", err.Error())
	}

	c, err := client.New(t)
	if err != nil {
		log.Fatalf("failed to create client, %s", err.Error())
	}
	log.Println("Created client, listening...")

	if err := c.StartReceiver(ctx, receive); err != nil {
		log.Fatalf("failed to start pubsub receiver, %s", err.Error())
	}
}

t, err := cepubsub.New(context.Background(),
	cepubsub.WithProjectID(env.ProjectID),
	cepubsub.WithSubscriptionID(env.SubscriptionID))
// ---中略---
c, err := client.New(t)

package main

import (
	"context"
	"log"
	"os"

	cloudevents "github.com/cloudevents/sdk-go"
	cepubsub "github.com/cloudevents/sdk-go/pkg/cloudevents/transport/pubsub"
	"github.com/kelseyhightower/envconfig"
)

type envConfig struct {
	ProjectID string `envconfig:"GOOGLE_CLOUD_PROJECT" required:"true"`
	TopicID   string `envconfig:"PUBSUB_TOPIC" default:"demo_cloudevents" required:"true"`
}

// Basic data struct.
type Model struct {
	Sequence int    `json:"id"`
	Message  string `json:"message"`
}

func main() {
	var env envConfig
	if err := envconfig.Process("", &env); err != nil {
		log.Fatalf("[ERROR] Failed to process env var: %s", err)
	}

	t, err := cepubsub.New(context.Background(),
		cepubsub.WithProjectID(env.ProjectID),
		cepubsub.WithTopicID(env.TopicID))
	if err != nil {
		log.Fatalf("failed to create pubsub transport, %s", err.Error())
	}
	c, err := cloudevents.NewClient(t, cloudevents.WithTimeNow(), cloudevents.WithUUIDs())
	if err != nil {
		log.Fatalf("failed to create client, %s", err.Error())
	}

	event := cloudevents.NewEvent(cloudevents.VersionV1)
	event.SetType("com.cloudevents.sample.sent")
	event.SetSource("github.com/cloudevents/sdk-go/cmd/samples/pubsub/sender/")
	_ = event.SetData(&Model{
		Sequence: 0,
		Message:  "HELLO",
	})

	_, _, err = c.Send(context.Background(), event)
	if err != nil {
		log.Fatalf("failed to send: %v", err)
	}
}

package receiver

import (
	"context"
	"encoding/json"
	"fmt"

	"cloud.google.com/go/pubsub"
	cepubsub "github.com/cloudevents/sdk-go/pkg/cloudevents/transport/pubsub"
)

// Model is published data struct.
type Model struct {
	Sequence int    `json:"id"`
	Message  string `json:"message"`
}

// Receiver parse payload to Model
func Receiver(ctx context.Context, msg *pubsub.Message) error {
	fmt.Printf("[INFO] message received: %v\n", msg)
	fmt.Printf("[INFO] message.Attributes: %s\n", msg.Attributes)
	fmt.Printf("[INFO] message.Data: %s\n", msg.Data)

	// convert pubsub.Message to cepubsub.Message
	var cemsg cepubsub.Message
	cemsg.Data = msg.Data
	cemsg.Attributes = msg.Attributes
	fmt.Printf("[INFO] CloudEventsVersion: %s\n", cemsg.CloudEventsVersion())

	// get data(type Model) from cepubsub.Message
	var data &Model
	json.Unmarshal([]byte(cemsg.Data), &data)
	fmt.Printf("[INFO] Model.Sequence: %d\n", data.Sequence)
	fmt.Printf("[INFO] Model.Message: %s\n", data.Message)

	return nil
}

type Message struct {
	// Data is the actual data in the message.
	Data []byte

	// Attributes represents the key-value pairs the current message
	// is labelled with.
	Attributes map[string]string
}

func (m Message) CloudEventsVersion() string {
	// Check as Binary encoding first.
	if m.Attributes != nil {
		// Binary v0.3:
		if s := m.Attributes[prefix+"specversion"]; s != "" {
			return s
		}
	}

	// Now check as Structured encoding.
	raw := make(map[string]json.RawMessage)
	if err := json.Unmarshal(m.Data, &raw); err != nil {
		return ""
	}

	// structured v0.3
	if v, ok := raw["specversion"]; ok {
		var version string
		if err := json.Unmarshal(v, &version); err != nil {
			return ""
		}
		return version
	}

	return ""
}