Documentation
¶
Overview ¶
Package broadcast pushes server events to browsers over named channels, in the spirit of Laravel Broadcasting: public, private ("private-") and presence ("presence-") channels, each authorized by a rule registered for its name pattern, fanned out in memory or through a Broker (drivers/redis) so every app instance delivers every event.
b := broadcast.New(broadcast.Config{
Broker: redisClient, // nil = this process only
User: func(ctx context.Context) (any, error) { return auth.UserFrom[*models.User](ctx, "web") },
Authorizer: gate,
})
err := b.Public("posts")
err = b.Private("user.{id}", func(ctx context.Context, u any, p map[string]string) (bool, error) {
return u.(*models.User).ID == p["id"], nil
})
r.Get("/broadcast/sse", b.SSE) // or drivers/websocket's Broadcast handler
err = b.Broadcast(ctx, PostPublished{Post: post}) // Channels() []string
Decisions fail closed: a channel without a matching rule, a rule that errors and a client without a user are all denied.
See https://github.com/AlexGalhardo/vukem/blob/main/docs/frontend.md.
Example ¶
package main
import (
"context"
"fmt"
"github.com/AlexGalhardo/vukem/broadcast"
)
type exPostPublished struct {
Title string `json:"title"`
}
func (exPostPublished) Channels() []string { return []string{"posts"} }
func main() {
b := broadcast.New(broadcast.Config{}) // Broker: redisClient across instances
defer b.Close()
if err := b.Public("posts"); err != nil {
fmt.Println(err)
return
}
// Browsers subscribe through r.Get("/broadcast/sse", b.SSE); here a
// subscription is opened directly.
ctx := context.Background()
sub, err := b.Subscribe(ctx, "posts")
if err != nil {
fmt.Println(err)
return
}
defer sub.Close()
if err := b.Broadcast(ctx, exPostPublished{Title: "Hello"}); err != nil {
fmt.Println(err)
return
}
msg := <-sub.Messages()
fmt.Println(msg.Channel, msg.Event, string(msg.Data))
// A channel without a rule is denied.
_, err = b.Subscribe(ctx, "private-user.1")
fmt.Println(err != nil)
}
Output: posts exPostPublished {"title":"Hello"} true
Index ¶
- Constants
- Variables
- func Listener[E Event](s *Server) events.Registration
- func ListenerAsync[E Event](s *Server) events.Registration
- func Queued[E Event](s *Server, kind string, opts ...queue.Option) events.Registration
- func ValidName(name string) bool
- type Config
- type Event
- type Member
- type Message
- type Server
- func (s *Server) Authorize(ctx context.Context, channel string) (Member, error)
- func (s *Server) Broadcast(ctx context.Context, event Event) error
- func (s *Server) Close()
- func (s *Server) MaxChannels() int
- func (s *Server) Presence(pattern string, ...) error
- func (s *Server) Private(pattern string, ...) error
- func (s *Server) PrivateGate(pattern, ability string, ...) error
- func (s *Server) Public(pattern string) error
- func (s *Server) Publish(ctx context.Context, channel, event string, data any) error
- func (s *Server) SSE(c *web.Ctx) error
- func (s *Server) SSEHandler() http.Handler
- func (s *Server) Subscribe(ctx context.Context, channels ...string) (*Subscription, error)
- type Subscription
Examples ¶
Constants ¶
const ( PrivatePrefix = "private-" PresencePrefix = "presence-" )
Channel name prefixes that select the channel kind.
const ( EventHere = "presence:here" EventJoining = "presence:joining" EventLeaving = "presence:leaving" )
Reserved event names for presence notifications.
const ( MaxNameLength = 128 DefaultMaxChannels = 32 )
Limits on names and subscriptions.
Variables ¶
var ( // ErrInvalidChannel rejects a malformed channel name (400). ErrInvalidChannel = &errs.HTTPError{Status: http.StatusBadRequest, Message: "Invalid channel name."} // ErrTooManyChannels rejects a subscription over Config.MaxChannels (400). ErrTooManyChannels = &errs.HTTPError{Status: http.StatusBadRequest, Message: "Too many channels."} // ErrForbidden denies a subscription (403); unauthenticated clients get // it too, so a probe cannot tell the two apart. ErrForbidden = &errs.HTTPError{Status: http.StatusForbidden, Message: "This channel is unauthorized."} )
Functions ¶
func Listener ¶
func Listener[E Event](s *Server) events.Registration
Listener broadcasts every dispatched E synchronously: Dispatch returns the publish error.
bus.Register(broadcast.Listener[PostPublished](b))
func ListenerAsync ¶
func ListenerAsync[E Event](s *Server) events.Registration
ListenerAsync broadcasts every dispatched E in the background; failures are logged by the bus.
Types ¶
type Config ¶
type Config struct {
// Broker carries events between instances; nil works in one process.
Broker sse.Broker
// User resolves the current user from a request context; nil or an
// error denies every private and presence channel.
User func(ctx context.Context) (any, error)
// Authorizer backs PrivateGate rules (authz.Gate implements it).
Authorizer web.Authorizer
// MaxChannels caps the channels of one connection (0 = 32).
MaxChannels int
}
Config configures a Server.
type Event ¶
type Event interface {
Channels() []string
}
Event is something broadcast on channels. The payload is BroadcastWith() when the event has it, the event itself otherwise, encoded as JSON; the event name is BroadcastAs() or the Go type name.
type Member ¶
Member is a user present on a presence channel. ID identifies the user (several connections of one user count once); Info is shown to the other members.
type Message ¶
type Message struct {
Channel string `json:"channel"`
Event string `json:"event"`
Data json.RawMessage `json:"data"`
}
Message is one event delivered to a subscriber.
type Server ¶
type Server struct {
// contains filtered or unexported fields
}
Server authorizes subscriptions and fans broadcast events out.
func (*Server) Authorize ¶
Authorize decides whether the client of ctx may subscribe to channel and returns its presence member. Errors are ErrInvalidChannel or ErrForbidden, with the cause wrapped for logs.
func (*Server) MaxChannels ¶
MaxChannels is the per-connection channel limit transports enforce.
func (*Server) Presence ¶
func (s *Server) Presence( pattern string, fn func(ctx context.Context, user any, params map[string]string) (Member, bool, error), ) error
Presence authorizes "presence-" channels matching pattern with fn, which returns the member the other subscribers see (its ID must not be empty).
func (*Server) Private ¶
func (s *Server) Private( pattern string, fn func(ctx context.Context, user any, params map[string]string) (bool, error), ) error
Private authorizes "private-" channels matching pattern (given without the prefix, e.g. "user.{id}") with fn, called with the current user and the pattern parameters.
func (*Server) PrivateGate ¶
func (s *Server) PrivateGate( pattern, ability string, load func(ctx context.Context, params map[string]string) (any, error), ) error
PrivateGate authorizes "private-" channels matching pattern through Config.Authorizer: ability is checked on the subject load returns for the parameters (load nil checks a gate ability with no subject).
func (*Server) Public ¶
Public declares public channels matching pattern ("posts", "posts.{id}"): anyone may subscribe. Undeclared public channels are denied, so a typo or a per-user name never becomes readable by default.
func (*Server) Publish ¶
Publish sends one event named event with data (encoded as JSON) on channel, for callers without an Event type.
func (*Server) SSE ¶
SSE is the server-sent events transport, a route handler for GET /broadcast/sse?channels=posts,private-user.42. Every channel is authorized before the stream starts; any refusal answers 400 or 403 and nothing is streamed. Each event is sent with its name as the SSE event type and the Message as JSON data; presence channels first get an EventHere event listing the members.
r.Get("/broadcast/sse", b.SSE)
func (*Server) SSEHandler ¶
SSEHandler is SSE for plain net/http muxes; refusals are written with http.Error.
type Subscription ¶
type Subscription struct {
// contains filtered or unexported fields
}
Subscription receives the events of authorized channels until Close, ctx cancellation, Server.Close, or until it falls too far behind (then Messages closes and the client should reconnect).
func (*Subscription) Channels ¶
func (sub *Subscription) Channels() []string
Channels lists the subscribed channels.
func (*Subscription) Close ¶
func (sub *Subscription) Close()
Close unsubscribes and announces leaving presence channels; it is safe to call more than once.
func (*Subscription) Members ¶
func (sub *Subscription) Members(channel string) ([]Member, bool)
Members lists the members present on a subscribed presence channel (this instance's view), or false for other channels.
func (*Subscription) Messages ¶
func (sub *Subscription) Messages() <-chan Message
Messages delivers the events; it closes when the subscription ends.
Source Files