sse

README

README for sse

This repository contains some code for Server-Sent Events (SSE).

The focus at first will be on providing an SSE client which is completely compliant to the spec, and more-or-less API equivalent to the corresponding EventSource API available in JavaScript.

The idea is that using this library as an SSE client in your code should result in server event implementations which are also useful from JavaScript clients, without excessive hacks.

Later, it may include functions for writing SSE server handlers which support things like retry negotation, keep-alives and resuming events by ID.

Current Status

Past proof-of-concept stage.

Client code: compliant protocol support is complete, minimal client in progress. From there, protocol features and specified API functionality like connection closing, event listeners and callback APIs will come in, in order of usefulness.

Server code: I have basic proof of concept code which is not in this repository yet, and will probably feature in the client test mocks first before the server API is designed and implemented.

See the TODO.org file for working notes.

Documentation

Index

• Constants
• Variables
• func SinkEvents(w http.ResponseWriter, code int, feed EventFeed, options ...ServerOption) error
• func SinkJSONEvents(w http.ResponseWriter, code int, feed AnyEventFeed) error
• func SplitFunc() func([]byte, bool) (int, []byte, error)
• type AnyEventFeed
• type ClientOption
• type Event
  • func (ev *Event) GetData() ([]byte, error)
  • func (ev *Event) Reader() io.ReadSeeker
• type EventFeed
  • func NewJSONEncoderFeed(anyEventFeed AnyEventFeed) EventFeed
• type EventSink
  • func NewEventSink(w http.ResponseWriter, feed EventFeed, options ...ServerOption) (*EventSink, error)
  • func (sink *EventSink) Respond(code int)
  • func (sink *EventSink) Sink() error
• type KeepAliveTime
  • func (kat KeepAliveTime) Apply(sink *EventSink) error
• type ReadyState
• type ReconnectTime
  • func (rt ReconnectTime) Apply(ssec *SSEClient) error
• type SSEClient
  • func NewSSEClient(options ...ClientOption) *SSEClient
  • func (ssec *SSEClient) Close()
  • func (ssec *SSEClient) Errors() <-chan error
  • func (ssec *SSEClient) GetStream(uri string) error
  • func (ssec *SSEClient) Messages() <-chan *Event
  • func (ssec *SSEClient) Opens() <-chan bool
  • func (ssec *SSEClient) Reopen()
  • func (ssec *SSEClient) SetContext(ctx context.Context)
  • func (ssec *SSEClient) URL() *url.URL
• type ServerOption
• type SinkEvent
• type StdLogger
• type WantFlag
  • func (wf WantFlag) Apply(ssec *SSEClient) error

Constants

const (
	// bitwise flags for selecting which events to pass through
	WantErrors    WantFlag = 1
	WantOpenClose WantFlag = 2
	WantMessages  WantFlag = 4

	// ReadyState is an enum representing the current status of the connection
	Connecting ReadyState = 0
	Open       ReadyState = 1
	Closed     ReadyState = 2
)

const (
	MessageType = "message"
	ErrorType   = "error"
)

message types/names

Variables

var (
	IDHeader    = []byte("id")
	EventHeader = []byte("event")
	DataHeader  = []byte("data")
	RetryHeader = []byte("retry")
)

protocol header constants

Functions

func SinkEvents

func SinkEvents(w http.ResponseWriter, code int, feed EventFeed, options ...ServerOption) error

SinkEvents is an more-or-less drop-in replacement for a responder in a net/http response handler. It handles all the SSE protocol for you - just feed it events.

func SinkJSONEvents

func SinkJSONEvents(w http.ResponseWriter, code int, feed AnyEventFeed) error

SinkJSONEvents is a wrapped-up handler for responding with anything that encoding/json can marshal, that you can happily `return` to in a net/http handler.

func SplitFunc

func SplitFunc() func([]byte, bool) (int, []byte, error)

SplitFunc returns a bufio.SplitFunc for the text/event-stream MIME type. This is a stateful scanner which tokenizes the event stream.

Types

type AnyEventFeed

type AnyEventFeed interface {
	GetEventChan(clientCloseChan <-chan struct{}) <-chan interface{}
}

AnyEventFeed represents an event feed which does not return events which have their own byte marshall method. Instead, encoding/json is used to marshal the events.

type ClientOption

type ClientOption interface {
	Apply(*SSEClient) error
}

type Event

type Event struct {
	// EventID is the ID of the event, or a previous event
	EventID string

	// Type is variously called "event type" and "event name" in the
	// TR.  Defaults to "message".  You must listen for specific named
	// event types to receive them.
	Type string

	// Error contains a go error if the error came from the client
	// (eg, connection problems)
	Error error

	// Data is the body of the event, and always terminated with a
	// line feed.  "Simple" events have this empty.  Returned as a
	// []byte as go decoders generally use that, but can't be binary!
	Data []byte

	// the RFC "Origin" field
	Origin string
}

Event is a structure holding SSE-compliant events

func (*Event) GetData

func (ev *Event) GetData() ([]byte, error)

GetData implements the SinkEvent interface, so that sse.Event can be used for sinks and sources.

func (*Event) Reader

func (ev *Event) Reader() io.ReadSeeker

Reader returns a reader for convenient passing to decoder functions.

type EventFeed

type EventFeed interface {
	GetEventChan(clientCloseChan <-chan struct{}) <-chan SinkEvent
}

EventFeed is a type for something that can return events in a form this API can write them to the write

func NewJSONEncoderFeed

func NewJSONEncoderFeed(anyEventFeed AnyEventFeed) EventFeed

NewJSONEncoderFeed converts an 'AnyEventFeed' to an EventFeed by marshalling each of the events returned via the EventFeed via encoding/json.

type EventSink

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

EventSink is a structure used by the event sink writer

func NewEventSink

func NewEventSink(w http.ResponseWriter, feed EventFeed, options ...ServerOption) (*EventSink, error)

NewEventSink returns an Event

func (*EventSink) Respond

func (sink *EventSink) Respond(code int)

Respond sets up the event channel - sends the HTTP headers, and starts writing a response. The keepalive timer is started.

func (*EventSink) Sink

func (sink *EventSink) Sink() error

Sink is the main event sink loop for the EventSink. Caller to provide the goroutine if required.

type KeepAliveTime

type KeepAliveTime time.Duration

func (KeepAliveTime) Apply

func (kat KeepAliveTime) Apply(sink *EventSink) error

type ReadyState

type ReadyState int32

type ReconnectTime

type ReconnectTime time.Duration

func (ReconnectTime) Apply

func (rt ReconnectTime) Apply(ssec *SSEClient) error

type SSEClient

type SSEClient struct {
	http.Client

	// all state changes are guarded by this
	sync.Mutex
	// contains filtered or unexported fields
}

SSEClient is a wrapper for an http Client which can also hold one SSE session open, and implements EventSource. You can use this API directly, but the EventSource API will steer you towards message protocols and behavior which also work from browsers.

func NewSSEClient

func NewSSEClient(options ...ClientOption) *SSEClient

NewSSEClient creates a new client which can make a single SSE call.

func (*SSEClient) Close

func (ssec *SSEClient) Close()

func (*SSEClient) Errors

func (ssec *SSEClient) Errors() <-chan error

Errors returns a channel from which errors will be returned. As an error indicates that the SSE channel is closed, there will only be one error returned before you reset the client (via Reopen())

func (*SSEClient) GetStream

func (ssec *SSEClient) GetStream(uri string) error

GetStream makes a GET request and returns a channel for *all* events read

func (*SSEClient) Messages

func (ssec *SSEClient) Messages() <-chan *Event

Messages returns a channel from which events can be read

func (*SSEClient) Opens

func (ssec *SSEClient) Opens() <-chan bool

func (*SSEClient) Reopen

func (ssec *SSEClient) Reopen()

Reopen allows a connection which was closed to be re-opened again.

func (*SSEClient) SetContext

func (ssec *SSEClient) SetContext(ctx context.Context)

SetContext allows the context to be specified - this affects cancelation and timeouts. Affects active client on reconnection only.

func (*SSEClient) URL

func (ssec *SSEClient) URL() *url.URL

URL returns the configured URL of the client

type ServerOption

type ServerOption interface {
	Apply(*EventSink) error
}

type SinkEvent

type SinkEvent interface {
	GetData() ([]byte, error)
}

SinkEvent is a generic type for things which can be marshalled to bytes. They might also implement any of the below interfaces to control behavior.

type StdLogger

type StdLogger interface {
	Print(v ...interface{})
	Printf(format string, v ...interface{})
	Println(v ...interface{})
	SetOutput(w io.Writer)
}

StdLogger is used to log error messages.

var Logger StdLogger = log.New(ioutil.Discard, "[sse] ", log.LstdFlags)

Logger is a way to enable logging for this module if you want it; approach borrowed with thanks from github.com/Shopify/sarama

type WantFlag

type WantFlag int32

func (WantFlag) Apply

func (wf WantFlag) Apply(ssec *SSEClient) error