client

package

v0.14.1 Latest Latest Go to latest Published: Aug 26, 2026 License: Apache-2.0 Imports: 11 Imported by: 2

Details

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

Repository

github.com/tsarna/vinculum-vws

Links

Open Source Insights

README ¶

WebSocket Client

This package provides a WebSocket client that implements the vinculum.Client interface, allowing you to connect to a vinculum WebSocket server and participate in pub/sub messaging with a single subscriber.

Features

Client Interface: Implements the vinculum.Client interface
Single Subscriber: Simplified design with one subscriber per client
Fluent Builder: Easy configuration using a fluent builder pattern
Graceful Connection: Proper connection lifecycle management
Topic Management: Clean subscription management with automatic server sync
Protocol Compliance: Follows the vinculum WebSocket protocol specification
Comprehensive Testing: Well-tested with unit tests and examples

Quick Start

package main

import (
    "context"
    "log"
    "time"

    "github.com/tsarna/vinculum/pkg/vinculum/vws/client"
    "go.uber.org/zap"
)

func main() {
    logger, _ := zap.NewDevelopment()

    // Create subscriber
    subscriber := &MySubscriber{}

    // Create client using fluent builder
    client, err := client.NewClient().
        WithURL("ws://localhost:8080/ws").
        WithLogger(logger).
        WithDialTimeout(10 * time.Second).
        WithSubscriber(subscriber).
        WithWriteChannelSize(200).  // Optional: configure buffer size
        WithAuthorization("Bearer your-token-here").  // Optional: add auth
        WithMonitor(&MyMonitor{}).  // Optional: add lifecycle monitoring
        Build()
    if err != nil {
        log.Fatal(err)
    }

    // Connect to server
    ctx := context.Background()
    if err := client.Connect(ctx); err != nil {
        log.Fatal(err)
    }
    defer client.Disconnect()

    // Use as Client
    client.Subscribe(ctx, "sensor/+/temperature")
    client.Publish(ctx, "sensor/room1/temperature", 23.5)
}

Builder Options

Required

WithURL(url string): WebSocket server URL (e.g., "ws://localhost:8080/ws")
WithSubscriber(subscriber vinculum.Subscriber): The subscriber that will receive events

Optional

WithLogger(logger *zap.Logger): Custom logger (defaults to nop logger)
WithDialTimeout(timeout time.Duration): Connection timeout (defaults to 30s)
WithWriteChannelSize(size int): Write channel buffer size (defaults to 100)
WithAuthorization(authHeader string): Static Authorization header value (convenience method)
WithAuthorizationProvider(provider AuthorizationProvider): Authorization provider function
WithHeaders(headers map[string][]string): Custom HTTP headers for WebSocket handshake
WithHeader(key, value string): Single HTTP header for WebSocket handshake (convenience method)
WithMonitor(monitor vinculum.ClientMonitor): Optional monitor for client lifecycle events

Client Interface

The client implements all vinculum.Client methods:

// Connection lifecycle
client.Connect(ctx)
client.Disconnect()

// Subscription management (simplified - no subscriber parameter needed)
client.Subscribe(ctx, topic)
client.Unsubscribe(ctx, topic) 
client.UnsubscribeAll(ctx)  // Sends UnsubscribeAll message to server

// Publishing
client.Publish(ctx, topic, payload)
client.PublishSync(ctx, topic, payload)  // Same as Publish for WebSocket

// Subscriber interface (delegates to configured subscriber)
client.OnSubscribe(ctx, topic)
client.OnUnsubscribe(ctx, topic)
client.OnEvent(ctx, topic, message, fields)
client.PassThrough(msg)

Subscription Management

The client provides simple subscription management:

Single Subscriber: One subscriber per client simplifies the design
No Local Tracking: Client doesn't track subscriptions locally - relies on server
Direct Server Communication: All subscribe/unsubscribe calls go directly to server

Protocol Support

The WebSocket server implements the Vinculum WebSocket Protocol, a JSON-based protocol.

Performance Tuning

Write Channel Buffer Size

The WithWriteChannelSize() option controls the internal buffer for outgoing messages:

Default (100): Good for most applications with moderate message rates
Larger Buffer (500-1000): Better for high-throughput applications that send many messages rapidly
Smaller Buffer (10-50): Lower memory usage for applications with infrequent messaging

// High-throughput configuration
client := client.NewClient().
    WithURL("ws://localhost:8080/ws").
    WithSubscriber(subscriber).
    WithWriteChannelSize(1000).  // Larger buffer for high message rates
    Build()

Trade-offs:

Larger buffers: Better performance under load, but use more memory
Smaller buffers: Lower memory usage, but may block on high message rates

Authorization

The client supports authorization for WebSocket connections using a provider function pattern.

Static Authorization (Convenience Method)

Use WithAuthorization() for fixed authorization headers:

client := client.NewClient().
    WithURL("ws://localhost:8080/ws").
    WithSubscriber(subscriber).
    WithAuthorization("Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...").
    Build()

This is a convenience method that internally creates a simple provider function.

Dynamic Authorization (Provider Function)

Use WithAuthorizationProvider() for tokens that need to be refreshed or computed:

// Example: JWT token refresh
tokenProvider := func(ctx context.Context) (string, error) {
    // Refresh token logic here
    token, err := refreshJWTToken(ctx)
    if err != nil {
        return "", err
    }
    return "Bearer " + token, nil
}

client := client.NewClient().
    WithURL("ws://localhost:8080/ws").
    WithSubscriber(subscriber).
    WithAuthorizationProvider(tokenProvider).
    Build()

Authorization Behavior

Error Handling: Provider errors prevent connection establishment
Context: Provider receives the dial context (respects timeouts)
Header Format: Authorization value is sent as-is in the Authorization header

Example Use Cases:

Static: API keys, long-lived tokens (use WithAuthorization())
Dynamic: JWT tokens, OAuth2 access tokens, rotating credentials (use WithAuthorizationProvider())

Client Monitoring

The client supports optional lifecycle monitoring through the vinculum.ClientMonitor interface:

Monitor Interface

type ClientMonitor interface {
    OnConnect(ctx context.Context)
    OnDisconnect(ctx context.Context, err error)
    OnSubscribe(ctx context.Context, topic string)
    OnUnsubscribe(ctx context.Context, topic string)
    OnUnsubscribeAll(ctx context.Context)
}

Examples

// Implement a custom monitor
type MyMonitor struct {
    logger *zap.Logger
}

func (m *MyMonitor) OnConnect(ctx context.Context) {
    m.logger.Info("Client connected")
}

func (m *MyMonitor) OnDisconnect(ctx context.Context, err error) {
    if err != nil {
        m.logger.Error("Client disconnected with error", zap.Error(err))
    } else {
        m.logger.Info("Client disconnected gracefully")
    }
}

func (m *MyMonitor) OnSubscribe(ctx context.Context, topic string) {
    m.logger.Info("Subscribed to topic", zap.String("topic", topic))
}

func (m *MyMonitor) OnUnsubscribe(ctx context.Context, topic string) {
    m.logger.Info("Unsubscribed from topic", zap.String("topic", topic))
}

func (m *MyMonitor) OnUnsubscribeAll(ctx context.Context) {
    m.logger.Info("Unsubscribed from all topics")
}

// Use the monitor
client, err := client.NewClient().
    WithURL("ws://localhost:8080/ws").
    WithSubscriber(subscriber).
    WithMonitor(&MyMonitor{logger: logger}).
    Build()

Event Details

OnConnect: Called after successful WebSocket connection establishment
OnDisconnect: Called when connection is closed
- err == nil: Graceful disconnect (via Disconnect() method)
- err != nil: Error-based disconnect (network issues, server errors, etc.)
OnSubscribe: Called after successful subscription to a topic
OnUnsubscribe: Called after successful unsubscription from a topic
OnUnsubscribeAll: Called after successful unsubscription from all topics

Use Cases

Logging: Track connection lifecycle and subscription changes
Metrics: Collect connection and subscription statistics
Alerting: Monitor for connection failures or unexpected disconnects
Debugging: Trace client behavior and troubleshoot issues
Health Checks: Monitor client connectivity status

Custom Headers

The client supports setting custom HTTP headers for the WebSocket handshake request.

Multiple Headers

Use WithHeaders() to set multiple headers at once:

client := client.NewClient().
    WithURL("ws://localhost:8080/ws").
    WithSubscriber(subscriber).
    WithHeaders(map[string][]string{
        "X-API-Key":    {"your-api-key"},
        "User-Agent":   {"MyApp/1.0"},
        "X-Client-ID":  {"client-123"},
        "Accept":       {"application/json"},
    }).
    Build()

Single Headers

Use WithHeader() for individual headers (convenience method):

client := client.NewClient().
    WithURL("ws://localhost:8080/ws").
    WithSubscriber(subscriber).
    WithHeader("X-API-Key", "your-api-key").
    WithHeader("User-Agent", "MyApp/1.0").
    WithHeader("X-Client-ID", "client-123").
    Build()

Headers with Authorization

Custom headers work alongside authorization:

client := client.NewClient().
    WithURL("ws://localhost:8080/ws").
    WithSubscriber(subscriber).
    WithHeaders(map[string][]string{
        "X-API-Key":  {"your-api-key"},
        "User-Agent": {"MyApp/1.0"},
    }).
    WithAuthorization("Bearer your-token").  // Authorization takes precedence
    Build()

Header Behavior

Merging: Multiple WithHeaders() calls merge headers together
Overriding: WithHeader() overwrites any existing header with the same name
Authorization Priority: WithAuthorization() and WithAuthorizationProvider() override any custom "Authorization" header
Case Sensitivity: Header names are case-sensitive (follow HTTP standards)

Example Use Cases:

API Keys: Custom authentication schemes
Client Identification: User-Agent, X-Client-ID headers
Content Negotiation: Accept, Accept-Language headers
Tracing: X-Trace-ID, X-Request-ID headers

Error Handling

The client handles various error conditions gracefully:

Connection Failures: Returns meaningful errors for connection issues
Protocol Errors: Handles server NACK responses appropriately
Network Issues: Detects and reports network problems
Graceful Shutdown: Safely closes connections and cleans up resources

Thread Safety

The client is designed to be thread-safe:

Concurrent Operations: Safe to call from multiple goroutines
Subscription Safety: Subscription management is properly synchronized
Connection Safety: Connection state is protected with proper locking

Testing

The package includes comprehensive tests:

Builder Tests: Validate configuration and defaults
Lifecycle Tests: Test connection management
Protocol Tests: Verify message handling (would require integration tests)
Error Tests: Validate error conditions

Run tests with:

go test ./pkg/vinculum/vws/client -v

Examples

See example_test.go for complete usage examples showing:

Basic client usage
Using client as a vinculum.Client
Subscriber implementation patterns

Documentation ¶

Examples ¶

Client
Client (AsClient)
Client (WithDynamicAuth)
Client (WithMonitor)

Constants ¶

This section is empty.

Variables ¶

This section is empty.

Functions ¶

This section is empty.

Types ¶

type AuthorizationProvider ¶

type AuthorizationProvider func(ctx context.Context) (string, error)

AuthorizationProvider is a function that returns an authorization header value. It receives a context and should return the authorization value (e.g., "Bearer token123") or an error if authorization cannot be obtained.

type Client ¶

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

Client implements the bus.Client interface over a WebSocket connection. It connects to a vinculum WebSocket server and provides pub/sub functionality.

Example ¶

ExampleClient demonstrates basic usage of the WebSocket client.

logger, _ := zap.NewDevelopment()

// Create a subscriber
subscriber := &exampleSubscriber{}

// Create client using fluent builder pattern
client, err := NewClient().
	WithURL("ws://localhost:8080/ws").
	WithLogger(logger).
	WithDialTimeout(10 * time.Second).
	WithSubscriber(subscriber).
	WithWriteChannelSize(200).                     // Configure write buffer size
	WithAuthorization("Bearer example-token-123"). // Add authorization
	Build()
if err != nil {
	log.Fatal(err)
}

// Connect to the server
ctx := context.Background()
if err := client.Connect(ctx); err != nil {
	log.Fatal(err)
}
defer client.Disconnect()

// Subscribe to topics
if err := client.Subscribe(ctx, "sensor/+/temperature"); err != nil {
	log.Fatal(err)
}

if err := client.Subscribe(ctx, "alerts/#"); err != nil {
	log.Fatal(err)
}

// Publish some events
client.Publish(ctx, "sensor/room1/temperature", 23.5)
client.Publish(ctx, "sensor/room2/temperature", 24.1)
client.Publish(ctx, "alerts/high-temperature", "Room 2 temperature is high")

// The subscriber will receive these events asynchronously
time.Sleep(100 * time.Millisecond)

Example (AsClient) ¶

ExampleClient_asClient demonstrates using the client as a bus.Client.

logger, _ := zap.NewDevelopment()

// Create a subscriber
subscriber := &exampleSubscriber{}

// Create client
client, err := NewClient().
	WithURL("ws://localhost:8080/ws").
	WithLogger(logger).
	WithSubscriber(subscriber).
	Build()
if err != nil {
	log.Fatal(err)
}

// Use Client interface methods
var vinculumClient bus.Client = client

// Connect to WebSocket
ctx := context.Background()
if err := vinculumClient.Connect(ctx); err != nil {
	log.Fatal(err)
}
defer vinculumClient.Disconnect()

// Now use it like any Client
vinculumClient.Subscribe(ctx, "notifications/#")
vinculumClient.Publish(ctx, "notifications/user/login", map[string]string{
	"user_id": "12345",
	"action":  "login",
})

time.Sleep(100 * time.Millisecond)

Example (WithDynamicAuth) ¶

ExampleClient_withDynamicAuth demonstrates using dynamic authorization.

logger, _ := zap.NewDevelopment()
subscriber := &exampleSubscriber{}

// Create a dynamic authorization provider
authProvider := func(ctx context.Context) (string, error) {
	// In a real application, this might:
	// - Refresh an expired JWT token
	// - Fetch a new OAuth2 access token
	// - Read credentials from a secure store

	// For this example, we'll simulate getting a fresh token
	token := "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.example"
	return "Bearer " + token, nil
}

// Create client with dynamic authorization
client, err := NewClient().
	WithURL("ws://localhost:8080/ws").
	WithLogger(logger).
	WithSubscriber(subscriber).
	WithAuthorizationProvider(authProvider). // Dynamic auth
	Build()
if err != nil {
	log.Fatal(err)
}

// Connect - authorization provider will be called during handshake
ctx := context.Background()
if err := client.Connect(ctx); err != nil {
	log.Fatal(err)
}
defer client.Disconnect()

// Use the client normally
client.Subscribe(ctx, "secure/data/#")
client.Publish(ctx, "secure/data/update", map[string]string{
	"message": "Authenticated message",
})

time.Sleep(100 * time.Millisecond)

Example (WithMonitor) ¶

ExampleClient_withMonitor demonstrates using a client monitor for lifecycle events.

logger, _ := zap.NewDevelopment()
subscriber := &exampleSubscriber{}

// Create a monitor to track client lifecycle events
monitor := &exampleMonitor{logger: logger}

// Create client with monitor
client, err := NewClient().
	WithURL("ws://localhost:8080/ws").
	WithLogger(logger).
	WithSubscriber(subscriber).
	WithMonitor(monitor). // Add lifecycle monitoring
	Build()
if err != nil {
	log.Fatal(err)
}

// Connect - monitor will receive OnConnect event
ctx := context.Background()
if err := client.Connect(ctx); err != nil {
	log.Fatal(err)
}

// Subscribe - monitor will receive OnSubscribe events
client.Subscribe(ctx, "events/#")
client.Subscribe(ctx, "alerts/+")

// Publish some events
client.Publish(ctx, "events/user/login", "user123")

// Unsubscribe - monitor will receive OnUnsubscribe event
client.Unsubscribe(ctx, "alerts/+")

// Disconnect - monitor will receive OnDisconnect event with nil error (graceful)
client.Disconnect()

time.Sleep(100 * time.Millisecond)

func (*Client) Connect ¶

func (c *Client) Connect(ctx context.Context) error

Connect establishes the WebSocket connection and starts message processing.

func (*Client) Disconnect ¶

func (c *Client) Disconnect() error

Disconnect closes the WebSocket connection and stops message processing.

func (*Client) IsConnected ¶ added in v0.14.0

func (c *Client) IsConnected() bool

IsConnected reports whether the client currently holds a live WebSocket connection: Connect has succeeded and neither Disconnect nor a read/write error has torn it down.

It is a snapshot, not a guarantee — the connection may drop between this call and the next Publish — which is what makes it useful for a health probe and useless as a precondition. Code that wants to publish should publish and handle the error.

It is deliberately stricter than the check the operational methods make: both the started flag and the connection itself must be present, so the brief window inside cleanup where the socket is already closed reads as disconnected rather than connected.

func (*Client) OnEvent ¶

func (c *Client) OnEvent(ctx context.Context, topic string, message any, fields map[string]string) error

OnEvent implements Subscriber.OnEvent

func (*Client) OnSubscribe ¶

func (c *Client) OnSubscribe(ctx context.Context, topic string) error

OnSubscribe implements Subscriber.OnSubscribe

func (*Client) OnUnsubscribe ¶

func (c *Client) OnUnsubscribe(ctx context.Context, topic string) error

OnUnsubscribe implements Subscriber.OnUnsubscribe

func (*Client) PassThrough ¶

func (c *Client) PassThrough(msg bus.EventBusMessage) error

PassThrough implements Subscriber.PassThrough

func (*Client) Publish ¶

func (c *Client) Publish(ctx context.Context, topic string, payload any) error

Publish implements Client.Publish

func (*Client) PublishSync ¶

func (c *Client) PublishSync(ctx context.Context, topic string, payload any) error

PublishSync implements Client.PublishSync - same as Publish for WebSocket client

func (*Client) SetSubscriber ¶ added in v0.9.1

func (c *Client) SetSubscriber(subscriber bus.Subscriber)

func (c *Client) Subscribe(ctx context.Context, topic string) error

Subscribe implements Client.Subscribe

func (*Client) Unsubscribe ¶

func (c *Client) Unsubscribe(ctx context.Context, topic string) error

Unsubscribe implements Client.Unsubscribe

func (*Client) UnsubscribeAll ¶

func (c *Client) UnsubscribeAll(ctx context.Context) error

UnsubscribeAll implements Client.UnsubscribeAll

type ClientBuilder ¶

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

ClientBuilder provides a fluent interface for building WebSocket clients.

func NewClient ¶

func NewClient() *ClientBuilder

NewClient creates a new WebSocket client builder.

func (*ClientBuilder) Build ¶

func (b *ClientBuilder) Build() (*Client, error)

Build creates and returns a new WebSocket client with the configured options.

func (*ClientBuilder) IsValid ¶

func (b *ClientBuilder) IsValid() error

IsValid checks that all required configuration is present.

func (*ClientBuilder) WithAuthorization ¶

func (b *ClientBuilder) WithAuthorization(authHeader string) *ClientBuilder

WithAuthorization sets a static Authorization header value. This will be sent with the WebSocket handshake request. Example: WithAuthorization("Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...")

func (*ClientBuilder) WithAuthorizationProvider ¶

func (b *ClientBuilder) WithAuthorizationProvider(provider AuthorizationProvider) *ClientBuilder

WithAuthorizationProvider sets an authorization provider function. This function will be called during connection to obtain the authorization header.

func (*ClientBuilder) WithDialTimeout ¶

func (b *ClientBuilder) WithDialTimeout(timeout time.Duration) *ClientBuilder

WithDialTimeout sets the timeout for establishing the WebSocket connection.

func (*ClientBuilder) WithHeader ¶

func (b *ClientBuilder) WithHeader(key, value string) *ClientBuilder

WithHeader sets a single HTTP header for the WebSocket handshake. This is a convenience method for setting individual headers. Example: WithHeader("X-API-Key", "key123")

func (*ClientBuilder) WithHeaders ¶

func (b *ClientBuilder) WithHeaders(headers map[string][]string) *ClientBuilder

WithHeaders sets custom HTTP headers for the WebSocket handshake. These headers will be sent along with the WebSocket upgrade request. Note: This will override any existing headers. Use multiple calls to add headers incrementally. Example: WithHeaders(map[string][]string{"X-API-Key": {"key123"}, "User-Agent": {"MyApp/1.0"}})

func (*ClientBuilder) WithLogger ¶

func (b *ClientBuilder) WithLogger(logger *zap.Logger) *ClientBuilder

WithLogger sets the logger for the client.

func (*ClientBuilder) WithMonitor ¶

func (b *ClientBuilder) WithMonitor(monitor bus.ClientMonitor) *ClientBuilder

WithMonitor sets an optional monitor that will receive client lifecycle events. The monitor will be called for connect, disconnect, subscribe, unsubscribe, and unsubscribe all events.

func (*ClientBuilder) WithSubscriber ¶

func (b *ClientBuilder) WithSubscriber(subscriber bus.Subscriber) *ClientBuilder

WithSubscriber sets the subscriber that will receive events from the client.

func (*ClientBuilder) WithURL ¶

func (b *ClientBuilder) WithURL(url string) *ClientBuilder

WithURL sets the WebSocket URL to connect to.

func (*ClientBuilder) WithWriteChannelSize ¶

func (b *ClientBuilder) WithWriteChannelSize(size int) *ClientBuilder

WithWriteChannelSize sets the buffer size for the internal write channel. A larger buffer allows more messages to be queued for writing, which can improve performance under high load but uses more memory. Default is 100.

Source Files ¶

View all Source files

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

README ¶

WebSocket Client

Features

Quick Start

Builder Options

Required

Optional

Client Interface

Subscription Management

Protocol Support

Performance Tuning

Write Channel Buffer Size

Authorization

Static Authorization (Convenience Method)

Dynamic Authorization (Provider Function)

Authorization Behavior

Client Monitoring

Monitor Interface

Examples

Event Details

Use Cases

Custom Headers

Multiple Headers

Single Headers

Headers with Authorization

Header Behavior

Error Handling

Thread Safety

Testing

Examples

Documentation ¶

Index ¶

Examples ¶

Constants ¶

Variables ¶

Functions ¶

Types ¶

type AuthorizationProvider ¶

type Client ¶

func (*Client) Connect ¶

func (*Client) Disconnect ¶

func (*Client) IsConnected ¶ added in v0.14.0

func (*Client) OnEvent ¶

func (*Client) OnSubscribe ¶

func (*Client) OnUnsubscribe ¶

func (*Client) PassThrough ¶

func (*Client) Publish ¶

func (*Client) PublishSync ¶

func (*Client) SetSubscriber ¶ added in v0.9.1

func (*Client) Subscribe ¶

func (*Client) Unsubscribe ¶

func (*Client) UnsubscribeAll ¶

type ClientBuilder ¶

func NewClient ¶

func (*ClientBuilder) Build ¶

func (*ClientBuilder) IsValid ¶

func (*ClientBuilder) WithAuthorization ¶

func (*ClientBuilder) WithAuthorizationProvider ¶

func (*ClientBuilder) WithDialTimeout ¶

func (*ClientBuilder) WithHeader ¶

func (*ClientBuilder) WithHeaders ¶

func (*ClientBuilder) WithLogger ¶

func (*ClientBuilder) WithMonitor ¶

func (*ClientBuilder) WithSubscriber ¶

func (*ClientBuilder) WithURL ¶

func (*ClientBuilder) WithWriteChannelSize ¶

Source Files ¶