package sse

import "github.com/r3labs/sse"

Index ¶

• Constants
• type Client
• func NewClient(url string) *Client
• func (c *Client) OnDisconnect(fn ConnCallback)
• func (c *Client) Subscribe(stream string, handler func(msg *Event)) error
• func (c *Client) SubscribeChan(stream string, ch chan *Event) error
• func (c *Client) SubscribeChanRaw(ch chan *Event) error
• func (c *Client) SubscribeChanRawWithContext(ctx context.Context, ch chan *Event) error
• func (c *Client) SubscribeChanWithContext(ctx context.Context, stream string, ch chan *Event) error
• func (c *Client) SubscribeRaw(handler func(msg *Event)) error
• func (c *Client) SubscribeRawWithContext(ctx context.Context, handler func(msg *Event)) error
• func (c *Client) SubscribeWithContext(ctx context.Context, stream string, handler func(msg *Event)) error
• func (c *Client) Unsubscribe(ch chan *Event)
• type ConnCallback
• type Event
• type EventLog
• func (e *EventLog) Add(ev *Event)
• func (e *EventLog) Clear()
• func (e *EventLog) Replay(s *Subscriber)
• type EventStreamReader
• func NewEventStreamReader(eventStream io.Reader) *EventStreamReader
• func (e *EventStreamReader) ReadEvent() ([]byte, error)
• type ResponseValidator
• type Server
• func New() *Server
• func (s *Server) Close()
• func (s *Server) CreateStream(id string) *Stream
• func (s *Server) HTTPHandler(w http.ResponseWriter, r *http.Request)
• func (s *Server) Publish(id string, event *Event)
• func (s *Server) RemoveStream(id string)
• func (s *Server) StreamExists(id string) bool
• type Stream
• type StreamRegistration
• type Subscriber

Constants ¶

const DefaultBufferSize = 1024

DefaultBufferSize size of the queue that holds the streams messages.

Types ¶

type Client ¶

type Client struct {
	URL        string
	Connection *http.Client
	Retry      time.Time

	Headers        map[string]string
	EncodingBase64 bool
	EventID        string

	ResponseValidator ResponseValidator
	ReconnectStrategy backoff.BackOff
	ReconnectNotify   backoff.Notify
	// contains filtered or unexported fields
}

Client handles an incoming server stream

func NewClient ¶

func NewClient(url string) *Client

NewClient creates a new client

func (*Client) OnDisconnect ¶

func (c *Client) OnDisconnect(fn ConnCallback)

OnDisconnect specifies the function to run when the connection disconnects

func (*Client) Subscribe ¶

func (c *Client) Subscribe(stream string, handler func(msg *Event)) error

Subscribe to a data stream

func (*Client) SubscribeChan ¶

func (c *Client) SubscribeChan(stream string, ch chan *Event) error

SubscribeChan sends all events to the provided channel

func (*Client) SubscribeChanRaw ¶

func (c *Client) SubscribeChanRaw(ch chan *Event) error

SubscribeChanRaw sends all events to the provided channel

func (*Client) SubscribeChanRawWithContext ¶

func (c *Client) SubscribeChanRawWithContext(ctx context.Context, ch chan *Event) error

SubscribeChanRawWithContext sends all events to the provided channel with context

func (*Client) SubscribeChanWithContext ¶

func (c *Client) SubscribeChanWithContext(ctx context.Context, stream string, ch chan *Event) error

SubscribeChanWithContext sends all events to the provided channel with context

func (*Client) SubscribeRaw ¶

func (c *Client) SubscribeRaw(handler func(msg *Event)) error

SubscribeRaw to an sse endpoint

func (*Client) SubscribeRawWithContext ¶

func (c *Client) SubscribeRawWithContext(ctx context.Context, handler func(msg *Event)) error

SubscribeRawWithContext to an sse endpoint with context

func (*Client) SubscribeWithContext ¶

func (c *Client) SubscribeWithContext(ctx context.Context, stream string, handler func(msg *Event)) error

SubscribeWithContext to a data stream with context

func (*Client) Unsubscribe ¶

func (c *Client) Unsubscribe(ch chan *Event)

Unsubscribe unsubscribes a channel

type ConnCallback ¶

type ConnCallback func(c *Client)

ConnCallback defines a function to be called on a particular connection event

type Event ¶

type Event struct {
	ID    []byte
	Data  []byte
	Event []byte
	Retry []byte
	// contains filtered or unexported fields
}

Event holds all of the event source fields

type EventLog ¶

type EventLog []*Event

EventLog holds all of previous events

func (*EventLog) Add ¶

func (e *EventLog) Add(ev *Event)

Add event to eventlog

func (*EventLog) Clear ¶

func (e *EventLog) Clear()

Clear events from eventlog

func (*EventLog) Replay ¶

func (e *EventLog) Replay(s *Subscriber)

Replay events to a subscriber

type EventStreamReader ¶

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

EventStreamReader scans an io.Reader looking for EventStream messages.

func NewEventStreamReader ¶

func NewEventStreamReader(eventStream io.Reader) *EventStreamReader

NewEventStreamReader creates an instance of EventStreamReader.

func (*EventStreamReader) ReadEvent ¶

func (e *EventStreamReader) ReadEvent() ([]byte, error)

ReadEvent scans the EventStream for events.

type ResponseValidator ¶

type ResponseValidator func(c *Client, resp *http.Response) error

ResponseValidator validates a response

type Server ¶

type Server struct {
	// Specifies the size of the message buffer for each stream
	BufferSize int
	// Enables creation of a stream when a client connects
	AutoStream bool
	// Enables automatic replay for each new subscriber that connects
	AutoReplay bool
	// Encodes all data as base64
	EncodeBase64 bool
	// Sets a ttl that prevents old events from being transmitted
	EventTTL time.Duration
	Streams  map[string]*Stream
	Headers  map[string]string
	// contains filtered or unexported fields
}

Server Is our main struct

func New ¶

func New() *Server

New will create a server and setup defaults

func (*Server) Close ¶

func (s *Server) Close()

Close shuts down the server, closes all of the streams and connections

func (*Server) CreateStream ¶

func (s *Server) CreateStream(id string) *Stream

CreateStream will create a new stream and register it

func (*Server) HTTPHandler ¶

func (s *Server) HTTPHandler(w http.ResponseWriter, r *http.Request)

HTTPHandler serves new connections with events for a given stream ...

func (*Server) Publish ¶

func (s *Server) Publish(id string, event *Event)

Publish sends a mesage to every client in a streamID

func (*Server) RemoveStream ¶

func (s *Server) RemoveStream(id string)

RemoveStream will remove a stream

func (*Server) StreamExists ¶

func (s *Server) StreamExists(id string) bool

StreamExists checks whether a stream by a given id exists

type Stream ¶

type Stream struct {
	// Enables replaying of eventlog to newly added subscribers
	AutoReplay bool
	Eventlog   EventLog
	// contains filtered or unexported fields
}

Stream ...

type StreamRegistration ¶

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

StreamRegistration ...

type Subscriber ¶

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

Subscriber ...

Source Files ¶

client.go event.go event_log.go http.go server.go stream.go subscriber.go