realtime

package
v1.0.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jul 3, 2026 License: MIT Imports: 15 Imported by: 0

Documentation

Overview

Package realtime is the OniWorks realtime platform — the nervous system of the framework. It provides a WebSocket connection manager, channel/room router, broadcast via Oni Memory, presence tracking, backpressure, auth per connection, and reconnect/resume support.

Index

Constants

View Source 
const (
	EventTypeConnect     = "oni:connect"
	EventTypeDisconnect  = "oni:disconnect"
	EventTypeError       = "oni:error"
	EventTypePing        = "oni:ping"
	EventTypePong        = "oni:pong"
	EventTypeResume      = "oni:resume"
	EventTypeAck         = "oni:ack"
	EventTypeSubscribe   = "oni:subscribe"
	EventTypeUnsubscribe = "oni:unsubscribe"
)

System event types (reserved — prefixed with "oni:")

Variables

This section is empty.

Functions

This section is empty.

Types

type Conn 

type Conn struct {

	// Authenticated user (0 = anonymous)
	UserID int64
	// contains filtered or unexported fields
}

Conn represents a single WebSocket connection to a client. It runs two goroutines: a read loop and a write loop. The send channel provides backpressure — slow clients are dropped after write timeout.

func (*Conn) Close 

func (c *Conn) Close()

Close terminates the connection gracefully.

func (*Conn) Get 

func (c *Conn) Get(key string) (any, bool)

Get retrieves a value from the per-connection metadata bag.

func (*Conn) ID 

func (c *Conn) ID() string

ID returns the unique connection identifier.

func (*Conn) IsSubscribed 

func (c *Conn) IsSubscribed(channel string) bool

IsSubscribed reports whether this connection is subscribed to a channel.

func (*Conn) Send 

func (c *Conn) Send(e *Event) bool

Send queues an event for delivery. Non-blocking — drops if buffer is full.

func (*Conn) Set 

func (c *Conn) Set(key string, value any)

Set stores a value in the per-connection metadata bag.

func (*Conn) Subscribe 

func (c *Conn) Subscribe(channel string)

Subscribe marks this connection as interested in a channel pattern. It also maintains the hub's channel→conns broadcast index so Broadcast only touches subscribers instead of scanning every connection on the node.

func (*Conn) Unsubscribe 

func (c *Conn) Unsubscribe(channel string)

Unsubscribe removes a channel subscription (and the broadcast-index entry).

type ConnectHandler 

type ConnectHandler func(c *Conn) error

ConnectHandler is called after a WebSocket connection is established and authenticated.

type DisconnectHandler 

type DisconnectHandler func(c *Conn)

DisconnectHandler is called when a WebSocket connection closes.

type Event 

type Event struct {
	// ID is a server-assigned monotonic event ID used for reconnect/resume.
	ID string `json:"id,omitempty"`

	// Type classifies the event (e.g. "chat.message", "presence.join").
	Type string `json:"type"`

	// Channel is the pub/sub channel this event belongs to.
	Channel string `json:"channel,omitempty"`

	// Payload is the event body — arbitrary JSON.
	Payload json.RawMessage `json:"payload,omitempty"`

	// Params are wildcard segments extracted by the channel router (not serialized).
	Params map[string]string `json:"-"`

	// ConnID is the sender's connection ID (not sent to clients).
	ConnID string `json:"-"`

	// UserID is the authenticated user's ID (0 = anonymous).
	UserID int64 `json:"-"`

	// Ts is a Unix timestamp in seconds.
	Ts int64 `json:"ts,omitempty"`
}

Event is the wire-format envelope for all messages flowing over a WebSocket connection.

JSON wire format: 

{"id":"01J...","type":"chat.message","channel":"chat.room1","payload":{...},"ts":1713000000}

func NewEvent 

func NewEvent(eventType, channel string, payload any) (*Event, error)

NewEvent creates an Event with an auto-generated ID and current timestamp.

func (*Event) Decode 

func (e *Event) Decode(dest any) error

Decode unmarshals the event payload into dest.

func (*Event) Encode 

func (e *Event) Encode() ([]byte, error)

Encode serializes the event to JSON bytes for transmission.

type EventBuffer 

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

EventBuffer stores recent events per channel for reconnect/resume. When a client reconnects with a last_event_id, the hub replays all events published after that ID — no messages lost during brief disconnects.

Rings are created on demand per channel and evicted again once idle for idleTTL (see janitor/evictIdle), so per-user or short-lived channels don't grow the buffer map without bound.

func NewEventBuffer 

func NewEventBuffer(maxSize int, maxAge time.Duration) *EventBuffer

NewEventBuffer creates a buffer retaining the last maxSize events per channel, up to maxAge old. Channel rings idle beyond the default idle TTL are eligible for eviction once a janitor is running (the Hub starts one automatically).

func (*EventBuffer) Push 

func (eb *EventBuffer) Push(channel string, e *Event)

Push appends an event to the channel's ring buffer.

lastTouched is stamped while eb.mu is held: evictIdle holds the write lock while it checks the stamp, so a ring being pushed to concurrently is always seen as fresh and can't be evicted out from under the push.

func (*EventBuffer) Since 

func (eb *EventBuffer) Since(channel, lastEventID string) []*Event

Since returns all buffered events for channel published after lastEventID. Pass "" to get all buffered events.

type HandlerFunc 

type HandlerFunc func(c *Conn, e *Event) error

HandlerFunc handles an incoming realtime event.

type Hub 

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

Hub is the central OniWorks realtime hub. It manages all WebSocket connections, routes events to channel handlers, broadcasts via Oni Memory (cross-node safe), and tracks presence.

func New 

func New(opts Options) *Hub

New creates a Hub with the given options.

func (*Hub) Broadcast 

func (h *Hub) Broadcast(channel string, payload any) error

Broadcast sends an event to all connections subscribed to channel, including connections on other nodes (via Oni Memory pub/sub). 

hub.Broadcast("chat.room1", myPayload)

func (*Hub) BroadcastEvent 

func (h *Hub) BroadcastEvent(channel string, e *Event) error

BroadcastEvent broadcasts a pre-built Event to a channel.

func (*Hub) Channel 

func (h *Hub) Channel(pattern string, fn HandlerFunc)

Channel registers a handler for events arriving on a channel pattern. 

hub.Channel("chat.{room}", func(c *Conn, e *Event) error {
    return hub.Broadcast("chat."+e.Params["room"], e.Payload)
})

func (*Hub) ConnCount 

func (h *Hub) ConnCount() int

ConnCount returns the number of active connections on this node.

func (*Hub) Handler 

func (h *Hub) Handler() onihttp.HandlerFunc

Handler returns an onihttp.HandlerFunc that upgrades HTTP connections to WebSocket. Mount it with: router.Get("/ws", hub.Handler())

func (*Hub) Members 

func (h *Hub) Members(channel string) []PresenceInfo

Members returns all presence info for a channel (cross-node via Oni Memory).

func (*Hub) OnConnect 

func (h *Hub) OnConnect(fn ConnectHandler)

OnConnect registers a hook called after every successful connection.

func (*Hub) OnDisconnect 

func (h *Hub) OnDisconnect(fn DisconnectHandler)

OnDisconnect registers a hook called when a connection closes.

func (*Hub) Presence 

func (h *Hub) Presence(pattern string)

Presence marks a channel pattern as a presence channel. When a client subscribes/unsubscribes, their entry is written to Oni Memory. 

hub.Presence("room.{id}")

func (*Hub) Push 

func (h *Hub) Push(connID string, e *Event) bool

Push sends an event directly to a specific connection by ID.

func (*Hub) RoomSize 

func (h *Hub) RoomSize(channel string) int

RoomSize returns the number of connections subscribed to a channel. Uses Oni Memory for cross-node accuracy.

func (*Hub) ServeHTTP 

func (h *Hub) ServeHTTP(w http.ResponseWriter, r *http.Request)

ServeHTTP upgrades an HTTP request to a WebSocket connection. Mount this on your router: 

router.Get("/ws", hub.ServeHTTP)

func (*Hub) Shutdown 

func (h *Hub) Shutdown()

Shutdown closes all connections and stops the hub.

type Options 

type Options struct {
	// Memory is the Oni Memory store used for cross-node broadcast and presence.
	Memory *memory.Store

	// CheckOrigin validates the WebSocket origin header. Return true to allow.
	// Defaults to allowing all origins (override in production).
	CheckOrigin func(r *http.Request) bool

	// AuthFunc authenticates a connection during the HTTP upgrade handshake.
	// Return the user ID (0 = anonymous) and an error to reject the connection.
	AuthFunc func(r *http.Request) (userID int64, err error)

	// EventBufferSize is the max number of events buffered per channel for resume.
	EventBufferSize int

	// EventBufferAge is how long events stay in the buffer (default: 2 minutes).
	EventBufferAge time.Duration

	// EventBufferIdleTTL is how long a channel's event ring may sit untouched
	// (no push, no resume read) before the janitor evicts it. Without eviction,
	// per-user channels grow the buffer map without bound. Default: 10 minutes.
	EventBufferIdleTTL time.Duration

	// PresenceChannels lists channel patterns that auto-track presence.
	PresenceChannels []string

	// Logger (defaults to slog.Default).
	Logger *slog.Logger
}

Options configures the Hub.

type PresenceInfo 

type PresenceInfo struct {
	UserID int64          `json:"user_id"`
	ConnID string         `json:"conn_id"`
	Meta   map[string]any `json:"meta,omitempty"`
}

PresenceInfo is the data stored per-member in a presence channel.

type PresenceManager 

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

PresenceManager manages who is "online" in each channel. Presence state is stored in Oni Memory so it is visible to all nodes.

func (*PresenceManager) Count 

func (pm *PresenceManager) Count(channel string) int

Count returns the number of online members in a channel.

func (*PresenceManager) Join 

func (pm *PresenceManager) Join(channel string, info PresenceInfo)

Join marks a user as present in a channel.

func (*PresenceManager) Leave 

func (pm *PresenceManager) Leave(channel string, connID string)

Leave removes a user from a presence channel.

func (*PresenceManager) LeaveAll 

func (pm *PresenceManager) LeaveAll(connID string)

LeaveAll removes all presence entries for a connection across all channels.

func (*PresenceManager) Members 

func (pm *PresenceManager) Members(channel string) []PresenceInfo

Members returns all currently online members of a channel.

func (*PresenceManager) Refresh 

func (pm *PresenceManager) Refresh(channel, connID string)

Refresh extends the TTL for a connection's presence (call periodically).

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.