cloudeventsbridge

Documentation

Index

• func NewFromEventBridgeEvent(jsonevent []byte) (ce.Event, error)
• func PutCloudEvent(ctx context.Context, client Client, bus string, event ce.Event) (string, error)
• type Client

Examples

• NewFromEventBridgeEvent
• PutCloudEvent

Functions

func NewFromEventBridgeEvent

func NewFromEventBridgeEvent(jsonevent []byte) (ce.Event, error)

Example

package main

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

	ceb "github.com/embano1/cloudeventsbridge"
)

func main() {
	handler := func(ctx context.Context, b json.RawMessage) {
		event, err := ceb.NewFromEventBridgeEvent(b)
		if err != nil {
			log.Fatalf("could not convert to cloudevent: %v", err)
		}
		fmt.Println(event)
	}

	handler(context.TODO(), []byte(lambdaInputEvent))
}

const lambdaInputEvent = `{
    "version": "0",
    "id": "20fbbf4b-d159-bc73-9741-1fb9af32761c",
    "detail-type": "order.created.event.v0",
    "source": "com.example.order.service",
    "account": "123456789",
    "time": "2022-10-31T13:19:40Z",
    "region": "us-east-1",
    "resources": [
        "ce:id:order-1",
        "ce:extension:correlationid:somecustomer",
		"arn:aws:dynamodb:us-west-1:123456789012:db-1/table-1"
    ],
    "detail": {
        "customerName": "somecustomer",
        "itemIDs": [
            "item-1",
            "item-2"
        ]
    }
}`

Output:

Context Attributes,
  specversion: 1.0
  type: order.created.event.v0
  source: com.example.order.service
  id: order-1
  time: 2022-10-31T13:19:40Z
  datacontenttype: application/json
Extensions,
  awsaccount: 123456789
  awsregion: us-east-1
  awsresources: arn:aws:dynamodb:us-west-1:123456789012:db-1/table-1
  correlationid: somecustomer
Data,
  {
    "customerName": "somecustomer",
    "itemIDs": [
      "item-1",
      "item-2"
    ]
  }

func PutCloudEvent

func PutCloudEvent(ctx context.Context, client Client, bus string, event ce.Event) (string, error)

TODO: func for batching/func opts,e.g. for trace header

Example

package main

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

	"github.com/aws/aws-sdk-go-v2/config"
	"github.com/aws/aws-sdk-go-v2/service/eventbridge"

	ce "github.com/cloudevents/sdk-go/v2"

	ceb "github.com/embano1/cloudeventsbridge"
)

const (
	defaultRegion = "us-east-1"
	defaultBus    = "default"
)

func main() {
	region := os.Getenv("AWS_DEFAULT_REGION")
	if region == "" {
		log.Printf("AWS_REGION variable not set, using default (%q)", defaultRegion)
		region = defaultRegion
	}

	cfg, err := config.LoadDefaultConfig(context.TODO(), config.WithRegion(region))
	if err != nil {
		log.Fatalf("could not load AWS SDK config: %v", err)
	}
	awsclient := eventbridge.NewFromConfig(cfg)

	e := ce.NewEvent()
	e.SetID("order-1")
	e.SetSource("com.example.order.service")
	e.SetType("order.created.event.v0")
	e.SetTime(time.Now().UTC())

	data := map[string]any{
		"customerName": "somecustomer",
		"itemIDs": []string{
			"item-1",
			"item-2",
		},
	}
	if err = e.SetData(ce.ApplicationJSON, data); err != nil {
		log.Fatalf("could not set cloudevent data: %v", err)
	}

	e.SetExtension("correlationid", "somecustomer")

	bus := os.Getenv("AWS_EVENT_BUS")
	if bus == "" {
		bus = defaultBus
	}

	log.Printf("sending cloudevent to event bus: %q (region: %q)", bus, region)
	id, err := ceb.PutCloudEvent(context.TODO(), awsclient, bus, e)
	if err != nil {
		log.Fatalf("could not send cloudevent: %v", err)
	}

	log.Printf("returned eventbridge event id: %s", id)

	// prints to stdout for test assertion below
	fmt.Println("successfully sent cloudevent")
}

Output:

successfully sent cloudevent

Types

type Client

type Client interface {
	PutEvents(ctx context.Context, params *eventbridge.PutEventsInput, optFns ...func(options *eventbridge.Options)) (*eventbridge.PutEventsOutput, error)
}