websocket

package module

v2.0.8 Latest Latest Go to latest Published: Jun 19, 2026 License: Apache-2.0 Imports: 22 Imported by: 0

Details

Valid go.mod file
Redistributable license
Tagged version
Stable version
Learn more about best practices

Repository

github.com/pucora/pucora-websocket

Links

Open Source Insights

README ¶

pucora-websocket

WebSocket gateway support for Pucora CE.

Implements RFC-6455 WebSocket proxying with two modes:

Multiplexing — many clients, one backend connection, JSON envelope protocol
Direct — transparent per-client backend connection

User documentation

See docs/websockets.md in the main repository for configuration, envelope format, JWT, and examples.

To publish this module as a standalone repository, see PUBLISHING.md.

Package layout

config.go       # Parse extra_config["websocket"]
envelope.go     # Multiplex JSON envelope encode/decode
session.go      # Per-client session and write pump
hub.go          # Multiplex hub, backend handshake, routing
direct.go       # Direct mode bidirectional relay
backoff.go      # Reconnect delay strategies
ping.go         # Client ping loop
metrics.go      # OpenTelemetry counters
transport.go    # Accept/dial options and read timeouts
helpers.go      # URL building, header propagation
hub_registry.go # Test-only hub reset helper
router/gin/     # HandlerFactory integration with Gin

Integration

Registered in pucora-ce via handler_factory.go:

handlerFactory = wsgin.HandlerFactory(logger, handlerFactory) // before JWT
handlerFactory = ginjose.HandlerFactory(handlerFactory, logger, rejecter)

JWT runs on the HTTP upgrade; the WebSocket handler runs after validation.

Dependencies

github.com/coder/websocket — WebSocket client/server
github.com/pucora/lura/v2 — endpoint config and Gin router types

Tests

go test ./...

From the CE repository root: make test-websocket or make check-fixtures.

License

Apache 2.0

Documentation ¶

Index ¶

Constants
Variables
func AcceptOptions(cfg Config) *websocket.AcceptOptions
func BackendWSURL(endpoint *config.EndpointConfig, c *gin.Context) (string, error)
func EncodeEnvelope(env Envelope) ([]byte, error)
func ExtractHeaders(c *gin.Context, allowed []string) http.Header
func ExtractParams(c *gin.Context) map[string]string
func IsConfigured(extra config.ExtraConfig) bool
func IsWebSocketUpgrade(r *http.Request) bool
func PingLoop(ctx context.Context, conn *websocket.Conn, period, pongWait time.Duration)
func ResetHubRegistry()
func RunDirect(ctx context.Context, client *websocket.Conn, backendURL string, ...)
func SessionFromParams(params map[string]string, uuid string) map[string]interface{}
type ClientSession
- func NewClientSession(id, url string, session map[string]interface{}, conn *websocket.Conn, ...) *ClientSession
type Config
- func Parse(extra config.ExtraConfig) (Config, error)
type Envelope
- func DecodeEnvelope(data []byte) (Envelope, error)
- func NewOutboundEnvelope(url string, session map[string]interface{}, payload []byte) Envelope
- func (e *Envelope) Payload() ([]byte, error)
type Hub
- func GetHub(endpoint string, cfg Config, logger logging.Logger) *Hub
- func (h *Hub) EnsureBackend(ctx context.Context, url string, headers http.Header) error
- func (h *Hub) HandleClient(ctx context.Context, s *ClientSession, endpointURL string)
- func (h *Hub) RegisterClient(s *ClientSession)
type Metrics
- func GetMetrics(disabled bool) *Metrics
- func (m *Metrics) ConnectionClosed(ctx context.Context, endpoint string)
- func (m *Metrics) ConnectionOpened(ctx context.Context, endpoint string)

Constants ¶

View Source

const Namespace = "websocket"

Variables ¶

View Source

var ErrNoConfig = errors.New("websocket: no config")

Functions ¶

func AcceptOptions ¶

func AcceptOptions(cfg Config) *websocket.AcceptOptions

AcceptOptions builds client accept options from config.

func BackendWSURL ¶

func BackendWSURL(endpoint *config.EndpointConfig, c *gin.Context) (string, error)

func EncodeEnvelope ¶

func EncodeEnvelope(env Envelope) ([]byte, error)

func ExtractHeaders ¶

func ExtractHeaders(c *gin.Context, allowed []string) http.Header

func ExtractParams ¶

func ExtractParams(c *gin.Context) map[string]string

func IsConfigured ¶

func IsConfigured(extra config.ExtraConfig) bool

func IsWebSocketUpgrade ¶

func IsWebSocketUpgrade(r *http.Request) bool

func PingLoop ¶

func PingLoop(ctx context.Context, conn *websocket.Conn, period, pongWait time.Duration)

func ResetHubRegistry ¶

func ResetHubRegistry()

ResetHubRegistry clears multiplex hubs between tests. It is intended for test isolation only.

func RunDirect ¶

func RunDirect(
	ctx context.Context,
	client *websocket.Conn,
	backendURL string,
	headers map[string][]string,
	cfg Config,
	logger logging.Logger,
	endpoint string,
	metrics *Metrics,
)

func SessionFromParams ¶

func SessionFromParams(params map[string]string, uuid string) map[string]interface{}

Types ¶

type ClientSession ¶

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

func NewClientSession ¶

func NewClientSession(id, url string, session map[string]interface{}, conn *websocket.Conn, buffer int) *ClientSession

type Config ¶

type Config struct {
	BackoffStrategy           string
	ConnectEvent              bool
	DisconnectEvent           bool
	DisableOTELMetrics        bool
	EnableDirectCommunication bool
	InputHeaders              []string
	MaxMessageSize            int64
	MaxRetries                int
	MessageBufferSize         int
	PingPeriod                time.Duration
	PongWait                  time.Duration
	ReadBufferSize            int
	WriteBufferSize           int
	ReturnErrorDetails        bool
	Subprotocols              []string
	Timeout                   time.Duration
	WriteWait                 time.Duration
}

func Parse ¶

func Parse(extra config.ExtraConfig) (Config, error)

type Envelope ¶

type Envelope struct {
	URL     string                 `json:"url,omitempty"`
	Session map[string]interface{} `json:"session,omitempty"`
	Body    string                 `json:"body,omitempty"`
}

func DecodeEnvelope ¶

func DecodeEnvelope(data []byte) (Envelope, error)

func NewOutboundEnvelope ¶

func NewOutboundEnvelope(url string, session map[string]interface{}, payload []byte) Envelope

func (*Envelope) Payload ¶

func (e *Envelope) Payload() ([]byte, error)

type Hub ¶

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

func GetHub ¶

func GetHub(endpoint string, cfg Config, logger logging.Logger) *Hub

func (*Hub) EnsureBackend ¶

func (h *Hub) EnsureBackend(ctx context.Context, url string, headers http.Header) error

func (*Hub) HandleClient ¶

func (h *Hub) HandleClient(ctx context.Context, s *ClientSession, endpointURL string)

func (*Hub) RegisterClient ¶

func (h *Hub) RegisterClient(s *ClientSession)

type Metrics ¶

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

func GetMetrics ¶

func GetMetrics(disabled bool) *Metrics

func (*Metrics) ConnectionClosed ¶

func (m *Metrics) ConnectionClosed(ctx context.Context, endpoint string)

func (*Metrics) ConnectionOpened ¶

func (m *Metrics) ConnectionOpened(ctx context.Context, endpoint string)

Source Files ¶

View all Source files

Directories ¶

Path	Synopsis
router
gin

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