nacre

Documentation

Index

• func NewRandString(src rand.Source, n int) string
• type AppConfig
• type ClientState
• type Config
  • func ParseConfig() (Config, error)
  • func (c Config) JSONString() string
  • func (c Config) String() string
• type HTTPServer
  • func NewHTTPServer(address string, hub Hub, rateLimiter RateLimiter) *HTTPServer
  • func (s *HTTPServer) Serve(ctx context.Context) error
  • func (s *HTTPServer) Shutdown(ctx context.Context) error
• type Hub
  • func NewRedisHub(client *redis.Client, maxRedisStreamLen int, ...) Hub
• type Peer
• type RateLimiter
  • func NewInMemoryRateLimiter() RateLimiter
• type RedisConfig
• type Root
  • func DefaultServer(cfg Config) (Root, error)
• type TCPServer
  • func NewTCPServer(address string, httpAddress string, hub Hub, rateLimiter RateLimiter) (*TCPServer, error)
  • func (s *TCPServer) Serve(ctx context.Context)

Functions

func NewRandString

func NewRandString(src rand.Source, n int) string

NewRandString returns a new string of length 'n' consisting of random alphabetical characters.

Kudos to @icza and @moorara for the code at: https://stackoverflow.com/a/31832326

Types

type AppConfig

type AppConfig struct {
	TCPAddr              string
	HTTPAddr             string
	BaseURL              string
	MaxRedisStreamLen    int
	MaxStreamPersistence time.Duration
}

AppConfig exposes Nacre-specific configuration options.

type ClientState

type ClientState string

ClientState indicates whether the data-streaming client is still connected.

const (
	ClientStateConnected    ClientState = "CONNECTED"
	ClientStateDisconnected ClientState = "DISCONNECTED"
	ClientStateUnknown      ClientState = "UNKNOWN"
)

Possible client states.

type Config

type Config struct {
	Redis RedisConfig
	App   AppConfig
}

Config is the root structure containing Nacre configuration.

func ParseConfig

func ParseConfig() (Config, error)

ParseConfig returns the default Nacre configuration merged with overridden configurations from environment variables.

func (Config) JSONString

func (c Config) JSONString() string

JSONString returns a JSON-like representation of the configuration.

func (Config) String

func (c Config) String() string

type HTTPServer

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

HTTPServer handles nacre's HTTP requests and websocket upgrades.

func NewHTTPServer

func NewHTTPServer(address string, hub Hub, rateLimiter RateLimiter) *HTTPServer

NewHTTPServer allocates a HTTP server for serving nacre's HTTP traffic.

func (*HTTPServer) Serve

func (s *HTTPServer) Serve(ctx context.Context) error

Serve HTTP traffic on the configured address.

func (*HTTPServer) Shutdown

func (s *HTTPServer) Shutdown(ctx context.Context) error

Shutdown delegates to the inner http.Server's shutdown function.

type Hub

type Hub interface {
	// FeedExists returns true if there is data for the identified feed.
	FeedExists(ctx context.Context, id string) (bool, error)
	// Push data to the identified feed.
	Push(ctx context.Context, id string, data []byte) error
	// Listen for data on the identified feed using the returned channel.
	Listen(ctx context.Context, id string) (<-chan []byte, error)
	// GetAll data entries for the identified feed.
	GetAll(ctx context.Context, id string) ([][]byte, error)

	// ClientState returns the current state of the client driving data to the identified feed.
	ClientState(ctx context.Context, id string) (ClientState, error)
	// ClientConnected updates the current state of the client to 'CONNECTED' for the identified feed.
	ClientConnected(ctx context.Context, id string) error
	// ClientConnected updates the current state of the client to 'DISCONNECTED' for the identified feed.
	ClientDisconnected(ctx context.Context, id string) error
}

Hub is a central client/peer and streamed data management layer.

func NewRedisHub

func NewRedisHub(client *redis.Client, maxRedisStreamLen int, maxStreamPersistenceDuration time.Duration) Hub

NewRedisHub allocates a new Redis-backed hub implementation.

type Peer

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

Peer represents a connected websocket peer and is responsible for driving the websocket read/write loops and peer connection maangement.

type RateLimiter

type RateLimiter interface {
	// Stop the rate limiter.
	Stop()
	// TryAddClient to the rate limiter, or return false if it can currently not be done.
	TryAddClient(ctx context.Context, ip string) bool
	// RemoveClient from the rate limiter.
	RemoveClient(ctx context.Context, ip string) bool
	// TryAddpeer to the rate limiter, or return false if it can currently not be done.
	TryAddPeer(ctx context.Context, id string) bool
	// RemovePeer from the rate limiter.
	RemovePeer(ctx context.Context, id string) bool
}

RateLimiter defines functions for keeping track of ongoing peer/client connections and limiting incoming request rates.

func NewInMemoryRateLimiter

func NewInMemoryRateLimiter() RateLimiter

NewInMemoryRateLimiter returns a memory-backed rate limiter for managing incoming peer/client requests. It also spins off a new background goroutine for garbage collection, which can be stopped with inMemoryRateLimiter.Stop().

Note that this implementation does _not_ manage distributed state. It only tracks local calls to the RateLimiter interface and can therefore not accurately rate limit in e.g. a horizontally-scaled deployment.

type RedisConfig

type RedisConfig struct {
	Host     string
	Port     string
	Password string
}

RedisConfig exposes Redis-specific configuration options.

type Root

type Root struct {
	Cfg Config

	Hub  Hub
	HTTP *HTTPServer
	TCP  *TCPServer
}

Root is the root struct defining the nacre server dependencies.

func DefaultServer

func DefaultServer(cfg Config) (Root, error)

DefaultServer returns a Root nacre instance with the default configuration and setup.

type TCPServer

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

TCPServer handles nacre's TCP clients and their data streams.

func NewTCPServer

func NewTCPServer(address string, httpAddress string, hub Hub, rateLimiter RateLimiter) (*TCPServer, error)

NewTCPServer returns a stoppable TCP server listening on the provided address.

func (*TCPServer) Serve

func (s *TCPServer) Serve(ctx context.Context)

Serve incoming TCP connections and handle them in new goroutines.