binding

Example of implementing a transport including a simple message type, and a transport sender and receiver.

package main

import (
	"context"
	"encoding/json"
	"io"

	"github.com/cloudevents/sdk-go/pkg/binding"
	"github.com/cloudevents/sdk-go/pkg/cloudevents"
	"github.com/cloudevents/sdk-go/pkg/cloudevents/transport"
)

// ExMessage is a json.RawMessage, which is just a byte slice
// containing a JSON encoded event.
type ExMessage struct{ json.RawMessage }

func (m ExMessage) Structured() (string, []byte) {
	return cloudevents.ApplicationCloudEventsJSON, []byte(m.RawMessage)
}

func (m ExMessage) Event() (e cloudevents.Event, err error) {
	err = json.Unmarshal(m.RawMessage, &e)
	return e, err
}

func (m ExMessage) Finish(error) {}

// ExSender sends by writing JSON encoded events to an io.Writer
type ExSender struct{ *json.Encoder }

func NewExSender(w io.Writer) ExSender { return ExSender{json.NewEncoder(w)} }

func (s ExSender) Send(ctx context.Context, m binding.Message) error {
	if t, b := m.Structured(); t != "" {
		// Fast case: if the Message is already structured JSON we can
		// send it directly, no need to decode and re-encode. Encoding a
		// json.RawMessage to a json.Encoder() is just a byte-buffer copy.
		return s.Encode(json.RawMessage(b))
	}
	// Some other message encoding. Decode as a generic cloudevents.Event
	// and then re-encode as JSON
	if e, err := m.Event(); err != nil {
		return err
	} else if err := s.Encode(e); err != nil {
		return err
	}
	return nil
}

// ExReceiver receives by reading JSON encoded events from an io.Reader
type ExReceiver struct{ *json.Decoder }

func NewExReceiver(r io.Reader) ExReceiver { return ExReceiver{json.NewDecoder(r)} }

func (sr ExReceiver) Receive(context.Context) (binding.Message, error) {
	var m ExMessage
	err := sr.Decode(&m) // This is just a byte copy since m is a json.RawMessage
	return m, err
}

// NewExTransport returns a transport.Transport which is implemented by
// an ExSender and an ExReceiver
func NewExTransport(r io.Reader, w io.Writer) transport.Transport {
	return binding.NewTransport(NewExSender(w), NewExReceiver(r))
}

// Example of implementing a transport including a simple message type,
// and a transport sender and receiver.
func main() {}

This example shows how to use a transport in sender, receiver, and intermediary processes.

The sender and receiver use the client.Client API to send and receive messages. the transport. Only the intermediary example actually uses the transport APIs for efficiency and reliability in forwarding events.

package main

import (
	"context"
	"fmt"
	"io"
	"strconv"

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

const count = 3 // Example ends after this many events.

// The sender uses the cloudevents.Client API, not the transport APIs directly.
func runSender(w io.Writer) error {
	c, err := client.New(NewExTransport(nil, w))
	if err != nil {
		return err
	}
	for i := 0; i < count; i++ {
		e := cloudevents.New()
		e.SetType("example.com/event")
		e.SetSource("example.com/source")
		e.SetID(strconv.Itoa(i))
		if err := e.SetData(fmt.Sprintf("hello %d", i)); err != nil {
			return err
		}
		if _, _, err := c.Send(context.TODO(), e); err != nil {
			return err
		}
	}
	return nil
}

// The receiver uses the cloudevents.Client API, not the transport APIs directly.
func runReceiver(r io.Reader) error {
	i := 0
	process := func(e cloudevents.Event) error {
		fmt.Printf("%s\n", e)
		i++
		if i == count {
			return io.EOF
		}
		return nil
	}
	c, err := client.New(NewExTransport(r, nil))
	if err != nil {
		return err
	}
	return c.StartReceiver(context.TODO(), process)
}

// The intermediary receives events and forwards them to another
// process using ExReceiver and ExSender directly.
//
// By forwarding a transport.Message instead of a cloudevents.Event,
// it allows the transports to avoid un-necessary decoding of
// structured events, and to exchange delivery status between reliable
// transports. Even transports using different protocols can ensure
// reliable delivery.
func runIntermediary(r io.Reader, w io.WriteCloser) error {
	defer w.Close()
	for {
		receiver := NewExReceiver(r)
		sender := NewExSender(w)
		for i := 0; i < count; i++ {
			if m, err := receiver.Receive(context.TODO()); err != nil {
				return err
			} else if err := sender.Send(context.TODO(), m); err != nil {
				return err
			}
		}
	}
}

// This example shows how to use a transport in sender, receiver,
// and intermediary processes.
//
// The sender and receiver use the client.Client API to send and
// receive messages.  the transport.  Only the intermediary example
// actually uses the transport APIs for efficiency and reliability in
// forwarding events.
func main() {
	r1, w1 := io.Pipe() // The sender-to-intermediary pipe
	r2, w2 := io.Pipe() // The intermediary-to-receiver pipe

	done := make(chan error)
	go func() { done <- runReceiver(r2) }()
	go func() { done <- runIntermediary(r1, w2) }()
	go func() { done <- runSender(w1) }()
	for i := 0; i < 2; i++ {
		if err := <-done; err != nil && err != io.EOF {
			fmt.Println(err)
		}
	}

}

Output:

Validation: valid
Context Attributes,
  specversion: 0.2
  type: example.com/event
  source: example.com/source
  id: 0
  contenttype: application/json
Data,
  "hello 0"

Validation: valid
Context Attributes,
  specversion: 0.2
  type: example.com/event
  source: example.com/source
  id: 1
  contenttype: application/json
Data,
  "hello 1"

Validation: valid
Context Attributes,
  specversion: 0.2
  type: example.com/event
  source: example.com/source
  id: 2
  contenttype: application/json
Data,
  "hello 2"