sse

Documentation

Index

• type AddressEvent
• type Event
• type MockServer
  • func (m *MockServer) BroadcastEvent(address string, event AddressEvent, data []byte)
  • func (m *MockServer) CreateStream(address string)
  • func (m *MockServer) HTTPHandler(w http.ResponseWriter, r *http.Request)
  • func (m *MockServer) StartPublishing() error
  • func (m *MockServer) StreamExists(address string) bool
• type Server
  • func (s *Server) BroadcastEvent(address string, event AddressEvent, data []byte)
  • func (s *Server) CreateStream(address string)
  • func (s *Server) HTTPHandler(w http.ResponseWriter, r *http.Request)
  • func (s *Server) StartPublishing() error
  • func (s *Server) StreamExists(address string) bool
• type ServerInterface
• type Storage

Types

type AddressEvent

type AddressEvent string

AddressEvent is an event sent to address SSE stream.

const (
	TransactionReceivedAddressEvent AddressEvent = "transaction_received"
	AccountCreatedAddressEvent      AddressEvent = "account_created"
	AccountCreditedAddressEvent     AddressEvent = "account_credited"
)

type Event

type Event struct {
	Address string       `db:"address"`
	Event   AddressEvent `db:"event"`
	Data    string       `db:"data"`
}

type MockServer

type MockServer struct {
	mock.Mock
}

MockServer is a mockable SSE server.

func (*MockServer) BroadcastEvent

func (m *MockServer) BroadcastEvent(address string, event AddressEvent, data []byte)

func (*MockServer) CreateStream

func (m *MockServer) CreateStream(address string)

func (*MockServer) HTTPHandler

func (m *MockServer) HTTPHandler(w http.ResponseWriter, r *http.Request)

func (*MockServer) StartPublishing

func (m *MockServer) StartPublishing() error

func (*MockServer) StreamExists

func (m *MockServer) StreamExists(address string) bool

type Server

type Server struct {
	Storage Storage `inject:""`
	// contains filtered or unexported fields
}

func (*Server) BroadcastEvent

func (s *Server) BroadcastEvent(address string, event AddressEvent, data []byte)

func (*Server) CreateStream

func (s *Server) CreateStream(address string)

func (*Server) HTTPHandler

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

func (*Server) StartPublishing

func (s *Server) StartPublishing() error

StartPublishing starts publishing events from the shared storage.

func (*Server) StreamExists

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

type ServerInterface

type ServerInterface interface {
	BroadcastEvent(address string, event AddressEvent, data []byte)
	StartPublishing() error
	CreateStream(address string)
	StreamExists(address string) bool
	HTTPHandler(w http.ResponseWriter, r *http.Request)
}

type Storage

type Storage interface {
	// AddEvent adds a new server-sent event to the storage.
	AddEvent(event Event) error
	// GetEventsSinceID returns all events since `id`. Used to load and publish
	// all broadcasted events.
	// It returns the last event ID, list of events or error.
	// If `id` is equal `-1`:
	//    * it should return the last event ID and empty list if at least one
	//      event has been broadcasted.
	//    * it should return 0 if no events have been broadcasted.
	GetEventsSinceID(id int64) (int64, []Event, error)
}

Storage contains history of sent events. Because each transaction and digitalbits account is always processed by a single Bifrost server, we need to broadcast events in case client streams events from the other Bifrost server.

It's used to broadcast events to all instances of Bifrost server and to handle clients' reconnections.