joaju

package module

Go to main page

Versions in this module

v0

v0.6.1

Aug 30, 2026

v0.6.0

Aug 27, 2026

v0.5.0

Aug 22, 2026

Changes in this version

+ type API struct

+ AppID string

+ Broker Broker

+ Connect ConnectPolicy

+ Log *slog.Logger

+ MaxBodySize int64

+ Registry Registry

+ Subscribe SubscriptionPolicy

+ type ChannelTally struct

+ Subscriptions int

+ Users map[string]bool

+ func (t ChannelTally) Members() []string

+ type FleetTally struct

+ Channels map[string]ChannelTally

+ Connections int

+ func (t FleetTally) Channel(requested string) ChannelTally

type Protocol

+ Routes func(api API) http.Handler

+ type Registry interface

+ Connections func(g auth.Grant) (int, error)

+ Fleet func(ctx context.Context, g auth.Grant, channel string) FleetTally

+ Terminate func(ctx context.Context, g auth.Grant, subject string) (int, error)

type Server

+ func (s *Server) Fleet(ctx context.Context, g auth.Grant, channel string) FleetTally

v0.4.0

Aug 20, 2026

Changes in this version

+ const ChannelNameCharacters

+ const EncryptedPrivateCacheChannelPrefix

+ const MaxChannelNameLength

+ var ErrChannelName = errors.New("joaju: the channel name is not one this protocol carries")

+ type Carrier interface

+ Carry func(ctx context.Context, e Event)

type Protocol

+ Refuse func(r Refusal) []byte

+ type Refusal uint8

+ const RefusalOverQuota

+ const RefusalRateLimited

+ const RefusalUnreadable

v0.3.0

Aug 16, 2026

v0.2.0

Aug 15, 2026

Changes in this version

type Broker

+ func NewMemoryBroker() Broker

+ func RelayedBroker(base Broker, relay *Relay) Broker

type Event

+ UserID string

type Protocol

+ func NewPusher(broker Broker, subscribe SubscriptionPolicy, cfg PusherConfig) Protocol

+ type PusherConfig struct

+ ActivityTimeout time.Duration

+ ClientEvents ClientEvents

+ Observer Observer

type Subscription

+ Auth string

+ ChannelData json.RawMessage

v0.1.0

Aug 15, 2026

Changes in this version

+ const CacheChannelPrefix

+ const ClientEventPrefix

+ const Connect

+ const DefaultMaxBodySize

+ const DefaultMaxConnections

+ const DefaultMaxMessageSize

+ const DefaultMetricsTimeout

+ const DefaultOutboundQueue

+ const DefaultPingInterval

+ const DefaultPongTimeout

+ const DefaultRetryInterval

+ const DefaultWriteTimeout

+ const EventCacheMiss

+ const EventConnectionEstablished

+ const EventError

+ const EventMemberAdded

+ const EventMemberRemoved

+ const EventPing

+ const EventPong

+ const EventSubscribe

+ const EventSubscriptionSucceeded

+ const EventUnsubscribe

+ const InternalPrefix

+ const MetricsTopic

+ const PresenceCacheChannelPrefix

+ const PrivateCacheChannelPrefix

+ const ProtocolPrefix

+ const ReasonClient

+ const ReasonLimit

+ const ReasonShutdown

+ const ReasonTerminated

+ const ReasonTimeout

+ const TopicPrefix

+ var ErrClientEventChannel = ProtocolError

+ var ErrClientEventsDisabled = ProtocolError

+ var ErrConnectionLimit = errors.New("joaju: this tenant is holding as many connections as it may")

+ var ErrInvalidMessage = ProtocolError

+ var ErrNoChannel = errors.New("joaju: no such channel")

+ var ErrNoGrant = fmt.Errorf("%w: joaju: no grant", auth.ErrForbidden)

+ var ErrNotSubscribed = ProtocolError

+ var ErrOriginNotAllowed = ProtocolError

+ var ErrOverQuota = ProtocolError

+ var ErrRateLimited = ProtocolError

+ var ErrRelayClosed = errors.New("joaju: the relay is closed")

+ var ErrSocketClosed = errors.New("joaju: the socket is closed")

+ var ErrUnauthorized = ProtocolError

+ var ErrWrongTenant = fmt.Errorf("%w: joaju: the grant and the channel belong to different tenants", ...)

+ func Encode(f Frame) ([]byte, error)

+ func MetricsReplyTopic(id InstanceID) string

+ func Topic(name ChannelName) string

+ type Broker interface

+ All func(ctx context.Context, g auth.Grant) ([]Channel, error)

+ Find func(ctx context.Context, g auth.Grant, name ChannelName) (Channel, error)

+ FindOrCreate func(ctx context.Context, g auth.Grant, name ChannelName) (Channel, error)

+ Remove func(ctx context.Context, g auth.Grant, name ChannelName) error

+ type Bus interface

+ Publish func(ctx context.Context, channel string, message any) (int64, error)

+ Subscribe func(ctx context.Context, channels []string, callback func(message, channel string)) error

+ type Channel interface

+ Broadcast func(ctx context.Context, e Event) error

+ BroadcastToAll func(ctx context.Context, e Event) error

+ Connections func() []Subscriber

+ Data func() map[string]any

+ Find func(id SocketID) (Subscriber, bool)

+ Name func() ChannelName

+ Subscribe func(ctx context.Context, g auth.Grant, conn *Connection, member Member) error

+ Subscribed func(conn *Connection) bool

+ Unsubscribe func(ctx context.Context, conn *Connection) error

+ func NewChannel(name ChannelName) (Channel, error)

+ type ChannelName struct

+ func NewChannelName(g auth.Grant, requested string) (ChannelName, error)

+ func (n ChannelName) IsZero() bool

+ func (n ChannelName) Requested() string

+ func (n ChannelName) String() string

+ func (n ChannelName) Tenant() string

+ func (n ChannelName) Type() ChannelType

+ type ChannelType uint8

+ const CacheChannel

+ const PresenceCacheChannel

+ const PresenceChannel

+ const PrivateCacheChannel

+ const PrivateChannel

+ const PublicChannel

+ func (t ChannelType) Cache() bool

+ func (t ChannelType) Guarded() bool

+ func (t ChannelType) Presence() bool

+ func (t ChannelType) String() string

+ type ClientEvents bool

+ const ClientEventsOff

+ const ClientEventsOn

+ func (c ClientEvents) Accept(f Frame, channel ChannelName, from SocketID, subscribed bool) (Event, error)

+ type ConnectPolicy = auth.Policy[Handshake]

+ type Connection struct

+ func NewConnection(g auth.Grant, id SocketID, sink Sink) (*Connection, error)

+ func (c *Connection) Grant() auth.Grant

+ func (c *Connection) ID() SocketID

+ func (c *Connection) Send(ctx context.Context, message []byte) error

+ func (c *Connection) Subject() auth.Subject

+ func (c *Connection) Tenant() string

+ func (c *Connection) Terminate(ctx context.Context) error

+ type Counter struct

+ func NewCounter() *Counter

+ func (c *Counter) ChannelCreated(_ context.Context, name ChannelName)

+ func (c *Counter) ChannelRemoved(_ context.Context, name ChannelName)

+ func (c *Counter) ConnectionClosed(_ context.Context, _ SocketID, tenant, _ string)

+ func (c *Counter) ConnectionOpened(_ context.Context, _ SocketID, tenant string)

+ func (c *Counter) MessageReceived(_ context.Context, _ SocketID, _ []byte)

+ func (c *Counter) MessageSent(_ context.Context, _ SocketID, _ []byte)

+ func (c *Counter) Read(tenant string) TenantCount

+ func (c *Counter) Tenants() []string

+ type ErrorCode int

+ const CodeInvalidMessage

+ const CodeOverCapacity

+ const CodeOverQuota

+ const CodeRateLimited

+ const CodeUnauthorized

+ type Event struct

+ Channel ChannelName

+ Data json.RawMessage

+ Name string

+ Socket SocketID

+ type Frame struct

+ Channel string

+ Data json.RawMessage

+ Event string

+ UserID string

+ func CacheMiss(name ChannelName) (Frame, error)

+ func ConnectionEstablished(id SocketID, activityTimeout time.Duration) Frame

+ func Decode(message []byte) (Frame, error)

+ func ErrorFrame(err error) Frame

+ func EventFrame(e Event) (Frame, error)

+ func MemberAdded(name ChannelName, m Member) (Frame, error)

+ func MemberRemoved(name ChannelName, m Member) (Frame, error)

+ func Ping() Frame

+ func Pong() Frame

+ func SubscriptionSucceeded(name ChannelName, data map[string]any) (Frame, error)

+ func (f *Frame) UnmarshalJSON(b []byte) error

+ func (f Frame) ClientMaySend() error

+ func (f Frame) IsClientEvent() bool

+ func (f Frame) IsInternal() bool

+ func (f Frame) IsProtocol() bool

+ func (f Frame) MarshalJSON() ([]byte, error)

+ func (f Frame) Subscribe() (SubscribeRequest, error)

+ func (f Frame) Unsubscribe() (string, error)

+ type Handshake struct

+ Origin string

+ Socket SocketID

+ type InstanceID string

+ type Member struct

+ Info json.RawMessage

+ UserID string

+ type NopObserver struct

+ func (NopObserver) ChannelCreated(context.Context, ChannelName)

+ func (NopObserver) ChannelRemoved(context.Context, ChannelName)

+ func (NopObserver) ConnectionClosed(context.Context, SocketID, string, string)

+ func (NopObserver) ConnectionOpened(context.Context, SocketID, string)

+ func (NopObserver) MessageReceived(context.Context, SocketID, []byte)

+ func (NopObserver) MessageSent(context.Context, SocketID, []byte)

+ type Observer interface

+ ChannelCreated func(ctx context.Context, name ChannelName)

+ ChannelRemoved func(ctx context.Context, name ChannelName)

+ ConnectionClosed func(ctx context.Context, id SocketID, tenant, reason string)

+ ConnectionOpened func(ctx context.Context, id SocketID, tenant string)

+ MessageReceived func(ctx context.Context, id SocketID, message []byte)

+ MessageSent func(ctx context.Context, id SocketID, message []byte)

+ type Protocol interface

+ Close func(ctx context.Context, conn *Connection)

+ Message func(ctx context.Context, conn *Connection, message []byte) error

+ Open func(ctx context.Context, conn *Connection) error

+ type ProtocolError struct

+ Code ErrorCode

+ Message string

+ func (e ProtocolError) Error() string

+ func (e ProtocolError) Frame() Frame

+ type Relay struct

+ func NewRelay(ctx context.Context, id InstanceID, bus Bus, log *slog.Logger) (*Relay, error)

+ func (r *Relay) Close() error

+ func (r *Relay) Degraded() bool

+ func (r *Relay) ID() InstanceID

+ func (r *Relay) Join(g auth.Grant, ch Channel) error

+ func (r *Relay) Joined() int

+ func (r *Relay) Leave(name ChannelName) error

+ func (r *Relay) Publish(ctx context.Context, e Event) error

+ type Server struct

+ func NewServer(cfg ServerConfig) (*Server, error)

+ func (s *Server) Close(ctx context.Context)

+ func (s *Server) Connections(g auth.Grant) (int, error)

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

+ func (s *Server) Terminate(ctx context.Context, g auth.Grant, subject string) (int, error)

+ type ServerConfig struct

+ AppID string

+ AppKey string

+ Broker Broker

+ Connect ConnectPolicy

+ Log *slog.Logger

+ MaxBodySize int64

+ MaxConnections int

+ MaxMessageSize int64

+ MaxMessagesPerSecond int

+ MetricsTimeout time.Duration

+ Observer Observer

+ OutboundQueue int

+ PingInterval time.Duration

+ PongTimeout time.Duration

+ Protocol Protocol

+ Relay *Relay

+ Subscribe SubscriptionPolicy

+ WriteTimeout time.Duration

+ type Sink interface

+ Send func(ctx context.Context, message []byte) error

+ Terminate func(ctx context.Context) error

+ type SocketID string

+ type SubscribeRequest struct

+ Auth string

+ Channel string

+ ChannelData json.RawMessage

+ func (r SubscribeRequest) Member() (Member, error)

+ type Subscriber struct

+ Conn *Connection

+ Member Member

+ type Subscription struct

+ Channel ChannelName

+ Member Member

+ Socket SocketID

+ type SubscriptionPolicy = auth.Policy[Subscription]

+ type TenantCount struct

+ Channels int64

+ Connections int64

+ Received int64

+ Sent int64

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL