sse

Documentation

Overview

This package contains the Server Sent Events implementation used by horizon.

Index

• Variables
• func Publish(topic string, blocking bool)
• func Subscribe(topic string) chan interface{}
• func Unsubscribe(notification chan interface{}, topic string)
• func WriteEvent(ctx context.Context, w http.ResponseWriter, e Event)
• func WritePreamble(ctx context.Context, w http.ResponseWriter) bool
• type Event
  • func (e Event) SseEvent() Event
• type Eventable
• type Stream
  • func NewStream(ctx context.Context, w http.ResponseWriter) *Stream
  • func (s *Stream) Done()
  • func (s *Stream) Err(err error)
  • func (s *Stream) Init()
  • func (s *Stream) IsDone() bool
  • func (s *Stream) Send(e Event)
  • func (s *Stream) SentCount() int
  • func (s *Stream) SetLimit(limit int)

Variables

var (
	ErrRateLimited = errors.New("Rate limit exceeded")
)

Functions

func Publish

func Publish(topic string, blocking bool)

Publish publishes to a PubSub subscription notification channel.

NOTE there is good reason to usually publish in a non-blocking manner i.e. skipping publishing and dropping sending the notification to the channel. The reason is in case channel queue is full, and there's already a notification waiting to be consumed by a subscription.

This can happen if multiple messages need to be published on short interval when sse.Execute() loop is still busy on acting on the previous action, and haven't fetched the next message yet.

It is OK to not publish a second message to the topic since the one already in the queue will trigger the action in the next sse.Execute() iteration.

Only reason to publish a notification in a blocking manner would be to write consistent unit tests where a subscription can wait for notification to be published in separate goroutine.

func Subscribe

func Subscribe(topic string) chan interface{}

Subscribe to topic by SSE connection, usually with an ID (account, ledger, tx). Once a change occurs in Horizon database happens, Publish() is called by ingestor so the subscription channel is notified.

func Unsubscribe

func Unsubscribe(notification chan interface{}, topic string)

Unsubscribe to a topic, for example when SSE connection is closed.

func WriteEvent

func WriteEvent(ctx context.Context, w http.ResponseWriter, e Event)

WriteEvent does the actual work of formatting an SSE compliant message sending it over the provided ResponseWriter and flushing.

func WritePreamble

func WritePreamble(ctx context.Context, w http.ResponseWriter) bool

WritePreamble prepares this http connection for streaming using Server Sent Events. It sends the initial http response with the appropriate headers to do so.

Types

type Event

type Event struct {
	Data  interface{}
	Error error
	ID    string
	Event string
	Retry int
}

Event is the packet of data that gets sent over the wire to a connected client.

func (Event) SseEvent

func (e Event) SseEvent() Event

SseEvent returns the SSE compatible form of the Event... itself.

type Eventable

type Eventable interface {
	// SseEvent returns the SSE compatible form of the implementer
	SseEvent() Event
}

Eventable represents an object that can be converted to an SSE compatible event.

type Stream

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

func NewStream

func NewStream(ctx context.Context, w http.ResponseWriter) *Stream

NewStream creates a new stream against the provided response writer.

func (*Stream) Done

func (s *Stream) Done()

func (*Stream) Err

func (s *Stream) Err(err error)

func (*Stream) Init

func (s *Stream) Init()

Init function is only executed once. It writes the preamble event which includes the HTTP response code and a hello message. This should be called before any method that writes to the client to ensure that the preamble has been sent first.

func (*Stream) IsDone

func (s *Stream) IsDone() bool

IsDone is safe to call concurrently and is exported.

func (*Stream) Send

func (s *Stream) Send(e Event)

func (*Stream) SentCount

func (s *Stream) SentCount() int

func (*Stream) SetLimit

func (s *Stream) SetLimit(limit int)