eventstreamapi

Documentation

Index

• Constants
• func ApplyHTTPTransportFixes(r *request.Request)
• func GetHeaderString(msg eventstream.Message, headerName string) (string, error)
• type BufferEncoder
  • func NewBufferEncoder() *BufferEncoder
  • func (e *BufferEncoder) Encode(msg eventstream.Message) ([]byte, error)
• type Encoder
• type EventReader
  • func NewEventReader(decoder *eventstream.Decoder, payloadUnmarshaler protocol.PayloadUnmarshaler, ...) *EventReader
  • func (r *EventReader) ReadEvent() (event interface{}, err error)
• type EventWriter
  • func NewEventWriter(encoder Encoder, pm protocol.PayloadMarshaler, ...) *EventWriter
  • func (w *EventWriter) WriteEvent(event Marshaler) error
• type Marshaler
• type OnceError
  • func NewOnceError() *OnceError
  • func (e *OnceError) Err() error
  • func (e *OnceError) ErrorSet() <-chan struct{}
  • func (e *OnceError) SetError(err error)
• type SignEncoder
  • func NewSignEncoder(signer StreamSigner, encoder Encoder) *SignEncoder
  • func (s *SignEncoder) Close() error
  • func (s *SignEncoder) Encode(msg eventstream.Message) error
• type StreamSigner
• type StreamWriter
  • func NewStreamWriter(eventWriter *EventWriter, streamCloser io.Closer) *StreamWriter
  • func (w *StreamWriter) Close() error
  • func (w *StreamWriter) Err() error
  • func (w *StreamWriter) ErrorSet() <-chan struct{}
  • func (w *StreamWriter) Send(ctx aws.Context, event Marshaler) error
• type UnknownMessageTypeError
  • func (e *UnknownMessageTypeError) Error() string
• type Unmarshaler

Constants

const (
	ChunkSignatureHeader = `:chunk-signature` // chunk signature for message
	DateHeader           = `:date`            // Date header for signature

	// Message header and values
	MessageTypeHeader    = `:message-type` // Identifies type of message.
	EventMessageType     = `event`
	ErrorMessageType     = `error`
	ExceptionMessageType = `exception`

	// Message Events
	EventTypeHeader = `:event-type` // Identifies message event type e.g. "Stats".

	// Message Error
	ErrorCodeHeader    = `:error-code`
	ErrorMessageHeader = `:error-message`

	// Message Exception
	ExceptionTypeHeader = `:exception-type`
)

EventStream headers with specific meaning to async API functionality.

const InputWriterCloseErrorCode = "EventStreamInputWriterCloseError"

InputWriterCloseErrorCode is used to denote an error occurred while closing the event stream input writer.

Functions

func ApplyHTTPTransportFixes

func ApplyHTTPTransportFixes(r *request.Request)

ApplyHTTPTransportFixes is a no-op for Go 1.18 and above.

func GetHeaderString

func GetHeaderString(msg eventstream.Message, headerName string) (string, error)

GetHeaderString returns the value of the header as a string. If the header is not set or the value is not a string an error will be returned.

Types

type BufferEncoder

type BufferEncoder struct {
	// contains filtered or unexported fields
}

BufferEncoder is a utility that provides a buffered event stream encoder

func NewBufferEncoder

func NewBufferEncoder() *BufferEncoder

NewBufferEncoder returns a new BufferEncoder initialized with a 1024 byte buffer.

func (*BufferEncoder) Encode

func (e *BufferEncoder) Encode(msg eventstream.Message) ([]byte, error)

Encode returns the encoded message as a byte slice. The returned byte slice will be modified on the next encode call and should not be held onto.

type Encoder

type Encoder interface {
	Encode(eventstream.Message) error
}

Encoder is an stream encoder that will encode an event stream message for the transport.

type EventReader

type EventReader struct {
	// contains filtered or unexported fields
}

EventReader provides reading from the EventStream of an reader.

func NewEventReader

func NewEventReader(
	decoder *eventstream.Decoder,
	payloadUnmarshaler protocol.PayloadUnmarshaler,
	unmarshalerForEventType func(string) (Unmarshaler, error),
) *EventReader

NewEventReader returns a EventReader built from the reader and unmarshaler provided. Use ReadStream method to start reading from the EventStream.

func (*EventReader) ReadEvent

func (r *EventReader) ReadEvent() (event interface{}, err error)

ReadEvent attempts to read a message from the EventStream and return the unmarshaled event value that the message is for.

For EventStream API errors check if the returned error satisfies the awserr.Error interface to get the error's Code and Message components.

EventUnmarshalers called with EventStream messages must take copies of the message's Payload. The payload will is reused between events read.

type EventWriter

type EventWriter struct {
	// contains filtered or unexported fields
}

EventWriter provides a wrapper around the underlying event stream encoder for an io.WriteCloser.

func NewEventWriter

func NewEventWriter(encoder Encoder, pm protocol.PayloadMarshaler, eventTypeFor func(Marshaler) (string, error),
) *EventWriter

NewEventWriter returns a new event stream writer, that will write to the writer provided. Use the WriteEvent method to write an event to the stream.

func (*EventWriter) WriteEvent

func (w *EventWriter) WriteEvent(event Marshaler) error

WriteEvent writes an event to the stream. Returns an error if the event fails to marshal into a message, or writing to the underlying writer fails.

type Marshaler

type Marshaler interface {
	MarshalEvent(protocol.PayloadMarshaler) (eventstream.Message, error)
}

Marshaler provides a marshaling interface for event types to event stream messages.

type OnceError

type OnceError struct {
	// contains filtered or unexported fields
}

OnceError wraps the behavior of recording an error once and signal on a channel when this has occurred. Signaling is done by closing of the channel.

Type is safe for concurrent usage.

func NewOnceError

func NewOnceError() *OnceError

NewOnceError return a new OnceError

func (*OnceError) Err

func (e *OnceError) Err() error

Err acquires a read-lock and returns an error if one has been set.

func (*OnceError) ErrorSet

func (e *OnceError) ErrorSet() <-chan struct{}

ErrorSet returns a channel that will be used to signal that an error has been set. This channel will be closed when the error value has been set for OnceError.

func (*OnceError) SetError

func (e *OnceError) SetError(err error)

SetError acquires a write-lock and will set the underlying error value if one has not been set.

type SignEncoder

type SignEncoder struct {
	// contains filtered or unexported fields
}

SignEncoder envelopes event stream messages into an event stream message payload with included signature headers using the provided signer and encoder.

func NewSignEncoder

func NewSignEncoder(signer StreamSigner, encoder Encoder) *SignEncoder

NewSignEncoder returns a new SignEncoder using the provided stream signer and event stream encoder.

func (*SignEncoder) Close

func (s *SignEncoder) Close() error

Close encodes a final event stream signing envelope with an empty event stream payload. This final end-frame is used to mark the conclusion of the stream.

func (*SignEncoder) Encode

func (s *SignEncoder) Encode(msg eventstream.Message) error

Encode takes the provided message and add envelopes the message with the required signature.

type StreamSigner

type StreamSigner interface {
	GetSignature(headers, payload []byte, date time.Time) ([]byte, error)
}

StreamSigner defines an interface for the implementation of signing of event stream payloads

type StreamWriter

type StreamWriter struct {
	// contains filtered or unexported fields
}

StreamWriter provides concurrent safe writing to an event stream.

func NewStreamWriter

func NewStreamWriter(eventWriter *EventWriter, streamCloser io.Closer) *StreamWriter

NewStreamWriter returns a StreamWriter for the event writer, and stream closer provided.

func (*StreamWriter) Close

func (w *StreamWriter) Close() error

Close terminates the writers ability to write new events to the stream. Any future call to Send will fail with an error.

func (*StreamWriter) Err

func (w *StreamWriter) Err() error

Err returns any error that occurred while attempting to write an event to the stream.

func (*StreamWriter) ErrorSet

func (w *StreamWriter) ErrorSet() <-chan struct{}

ErrorSet returns a channel which will be closed if an error occurs.

func (*StreamWriter) Send

func (w *StreamWriter) Send(ctx aws.Context, event Marshaler) error

Send writes a single event to the stream returning an error if the write failed.

Send may be called concurrently. Events will be written to the stream safely.

type UnknownMessageTypeError

type UnknownMessageTypeError struct {
	Type    string
	Message eventstream.Message
}

UnknownMessageTypeError provides an error when a message is received from the stream, but the reader is unable to determine what kind of message it is.

func (*UnknownMessageTypeError) Error

func (e *UnknownMessageTypeError) Error() string

type Unmarshaler

type Unmarshaler interface {
	UnmarshalEvent(protocol.PayloadUnmarshaler, eventstream.Message) error
}

Unmarshaler provides the interface for unmarshaling a EventStream message into a SDK type.