reconnect

package
v0.8.2 Latest Latest
Warning

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

 Go to latest Published: Jul 19, 2026 License: MIT Imports: 22 Imported by: 2

README

Generic Reconnecting Network Client

The reconnect package provides a robust, generic network client with automatic reconnection capabilities for both TCP and Unix domain sockets. It implements retry mechanisms, connection lifecycle management, and configurable callbacks for handling connection events.

Overview

The reconnecting network client automatically handles connection failures by implementing configurable retry logic. It supports both TCP connections and Unix domain sockets, provides hooks for custom connection handling, session management, and error processing whilst maintaining thread safety and proper resource clean-up.

Key Features

  • Automatic Reconnection: Transparent reconnection with configurable retry strategies.
  • Multiple Network Types: Support for both TCP and Unix domain sockets.
  • Lifecycle Callbacks: Customisable hooks for socket, connect, session, disconnect, and error events.
  • Thread Safety: Built-in synchronisation for concurrent operations.
  • Timeout Management: Separate dial, read, and write timeout configuration.
  • Structured Logging: Integration with darvaza.org/slog for comprehensive logging.
  • Context Support: Full context.Context integration for cancellation and deadlines.
  • Configuration Validation: Built-in validation and default-setting mechanisms.

Basic Usage

import (
    "context"
    "time"

    "darvaza.org/x/net/reconnect"
)

// Get a logger instance (implementation-specific).
// This could be from any slog handler (filter, zap, etc.)
// or a custom implementation.
logger := getLogger()

// Create a configuration.
cfg := &reconnect.Config{
    Context: context.Background(),
    Logger:  logger,

    // Connection settings.
    // TCP: "host:port" or Unix: "/path/to/socket" or "unix:/path"
    Remote:       "example.com:8080",
    KeepAlive:    5 * time.Second,  // Default: 5s.
    DialTimeout:  2 * time.Second,  // Default: 2s.
    ReadTimeout:  2 * time.Second,  // Default: 2s.
    WriteTimeout: 2 * time.Second,  // Default: 2s.

    // Retry configuration.
    ReconnectDelay: time.Second,  // Delay between reconnection attempts.
}

// Create client.
client, err := reconnect.New(cfg)
if err != nil {
    logger.Fatal().Printf("Failed to create client: %v", err)
}

// Connect and start the client.
if err := client.Connect(); err != nil {
    logger.Fatal().Printf("Failed to connect: %v", err)
}

// Wait for completion.
defer func() {
    if err := client.Wait(); err != nil {
        logger.Error().Printf("Client error: %v", err)
    }
}()
Unix Domain Socket Usage

The client automatically detects Unix domain sockets based on the Remote string:

// Unix socket with explicit prefix
cfg := &reconnect.Config{
    Remote: "unix:/var/run/app.sock",
    // ... other configuration
}

// Unix socket with absolute path (auto-detected)
cfg := &reconnect.Config{
    Remote: "/var/run/app.sock",
    // ... other configuration
}

// Unix socket with .sock extension (auto-detected)
cfg := &reconnect.Config{
    Remote: "./app.sock",
    // ... other configuration
}

// Abstract Unix socket with @ prefix (Linux, auto-detected)
cfg := &reconnect.Config{
    Remote: "@app",
    // ... other configuration
}

Advanced Configuration

The client supports extensive customisation through callback functions and options.

Connection Lifecycle Callbacks
import (
    "context"
    "net"
    "syscall"

    "darvaza.org/slog"
    "darvaza.org/x/net/reconnect"
)

// Assume logger is already obtained.
var logger slog.Logger

cfg := &reconnect.Config{
    // Called against the raw socket before connecting.
    OnSocket: func(ctx context.Context, conn syscall.RawConn) error {
        // Configure socket options.
        return nil
    },

    // Called when a connection is established.
    OnConnect: func(ctx context.Context, conn net.Conn) error {
        logger.Info().Printf("Connected to %s", conn.RemoteAddr())
        // Perform handshake or initial setup.
        return nil
    },

    // Called for each session after connection.
    OnSession: func(ctx context.Context) error {
        // Implement your protocol logic here.
        // This function should block until the session ends.
        return handleSession(ctx)
    },

    // Called after the connection has been closed.
    OnDisconnect: func(ctx context.Context, conn net.Conn) error {
        logger.Info().Printf("Disconnected from %s", conn.RemoteAddr())
        // Clean up connection-specific resources.
        return nil
    },

    // Called when errors occur. The returned error replaces the
    // original for the reconnection logic.
    OnError: func(ctx context.Context, conn net.Conn, err error) error {
        logger.Error().
            WithField("error", err).
            Printf("Connection error")
        // Return nil to discard the error and keep retrying,
        // or reconnect.ErrDoNotReconnect to stop the client.
        return nil
    },
}
Retry Configuration
import (
    "context"
    "time"

    "darvaza.org/x/net/reconnect"
)

cfg := &reconnect.Config{
    // Simple constant delay between retries.
    ReconnectDelay: time.Second,

    // Custom wait function for retry logic.
    WaitReconnect: func(ctx context.Context) error {
        // Implement custom backoff logic.
        // Return error to stop reconnection.
        return customBackoff(ctx)
    },
}

// Use helper functions for common patterns.
cfg.WaitReconnect = reconnect.NewConstantWaiter(5 * time.Second)

// Immediate error return (no retry).
cfg.WaitReconnect = reconnect.NewImmediateErrorWaiter(errNoRetry)

// Prevent all reconnection attempts.
cfg.WaitReconnect = reconnect.NewDoNotReconnectWaiter(errStop)

Configuration Options

The Config structure supports the following fields:

Field Type Default Description
Context context.Context context.Background() Base context for the client.
Logger slog.Logger Default logger Logger for structured logging.
Remote string Required Target address. TCP: host:port, Unix: /path/to/socket or unix:/path
KeepAlive time.Duration 5s TCP keep-alive interval (ignored for Unix sockets).
DialTimeout time.Duration 2s Connection establishment timeout.
ReadTimeout time.Duration 2s Default read deadline, applied via the Reset*Deadline helpers, not automatically.
WriteTimeout time.Duration 2s Default write deadline, applied via the Reset*Deadline helpers, not automatically.
ReconnectDelay time.Duration 0 Delay between reconnection attempts. Zero means 5s (DefaultWaitReconnect); negative disables reconnection.
WaitReconnect Waiter NewConstantWaiter(ReconnectDelay) Custom reconnection wait function.
OnSocket func nil Raw socket configuration callback.
OnConnect func nil Connection establishment callback.
OnSession func nil Session handler (blocks until done).
OnDisconnect func nil Disconnection callback.
OnError func nil Error handler callback.

Client Methods

Core Methods
// New creates a new Client with options.
func New(cfg *Config, options ...OptionFunc) (*Client, error)

// Connect starts the reconnection loop. A nil return doesn't mean
// a connection is established, only that the loop has started.
func (c *Client) Connect() error

// Config returns the configuration object.
func (c *Client) Config() *Config

// Reload attempts to apply configuration changes.
// Note: Currently returns ErrTODO.
func (c *Client) Reload() error

// Wait blocks until the client stops and returns the cancellation
// reason, nil when the shutdown was user-initiated.
func (c *Client) Wait() error

// Err returns the cancellation reason.
func (c *Client) Err() error

// Done returns a channel that closes once the client has stopped.
func (c *Client) Done() <-chan struct{}

// Shutdown initiates a shutdown and waits until done or context timeout.
func (c *Client) Shutdown(ctx context.Context) error
Configuration Methods
// SetDefaults fills gaps in the configuration.
func (cfg *Config) SetDefaults() error

// Valid checks if the configuration is usable.
func (cfg *Config) Valid() error

// ExportDialer creates a net.Dialer from the configuration.
func (cfg *Config) ExportDialer() net.Dialer
Utility Functions
// ParseRemote determines network type and address from a remote string.
// Returns (network, address, error) where network is reconnect.NetworkTCP or
// reconnect.NetworkUnix.
func ParseRemote(remote string) (network, address string, err error)

// ValidateRemote validates a remote address for TCP or Unix socket connection.
// Returns nil if the address is valid for either protocol.
func ValidateRemote(remote string) error

// TimeoutToAbsoluteTime adds duration to base time.
// Returns zero time if duration is negative.
func TimeoutToAbsoluteTime(base time.Time, d time.Duration) time.Time
Using Utility Functions
import "darvaza.org/x/net/reconnect"

// Parse and validate remote addresses
network, address, err := reconnect.ParseRemote("unix:/var/run/app.sock")
if err != nil {
    // Handle parsing error
}
// network = "unix", address = "/var/run/app.sock"

// Validate before using in configuration
if err := reconnect.ValidateRemote("example.com:8080"); err != nil {
    // Handle invalid address
}

// Convert relative timeout to absolute time
deadline := reconnect.TimeoutToAbsoluteTime(time.Now(), 30*time.Second)

Error Handling

The client distinguishes between recoverable and non-recoverable errors.

Error Types
  • ErrConfigBusy: the Config is already in use by another client.
  • ErrRunning: the client has already been started.
  • ErrDoNotReconnect: instructs the client to stop reconnecting.
  • ErrClosed: the client or stream session has already been shut down. It wraps darvaza.org/x/sync/errors.ErrClosed, the workgroup's sentinel.
  • ErrNotConnected: the client isn't currently connected. It wraps ErrClosed, so one errors.Is target covers both.

See errors.go for the full list.

Error Classification
  • Fatal: ErrDoNotReconnect, possibly wrapped, terminates the client.
  • Context termination: the client stops retrying once its context is cancelled or its deadline expires.
  • Recoverable: everything else — connection refused/reset/aborted, timeouts, and other session errors — leads to a reconnection attempt.

The OnError callback observes every error, and its return value replaces it: return nil to discard the error, or ErrDoNotReconnect to stop the client.

Thread Safety

All client operations are thread-safe. Multiple goroutines can safely:

  • Call client methods concurrently.
  • Access the client's context.
  • Trigger cancellation.

The configuration becomes immutable after creating a client. Attempting to reuse a configuration for another client returns ErrConfigBusy.

Resource Management

The client properly manages resources:

  • Connections are closed at the end of each session.
  • Goroutines are cleaned up on shutdown.
  • Context cancellation is propagated to callbacks and workers.
  • The Wait() method ensures proper shutdown sequencing.

Helper Functions

Waiter Functions
// NewConstantWaiter creates a waiter with fixed delay.
func NewConstantWaiter(d time.Duration) func(context.Context) error

// NewImmediateErrorWaiter returns an error immediately.
func NewImmediateErrorWaiter(err error) func(context.Context) error

// NewDoNotReconnectWaiter prevents reconnection.
func NewDoNotReconnectWaiter(err error) func(context.Context) error
Worker Functions
// NewShutdownFunc creates a worker that shuts down gracefully.
func NewShutdownFunc(s Shutdowner, timeout time.Duration) WorkerFunc

// NewCatchFunc creates an error catcher with exceptions.
func NewCatchFunc(nonErrors ...error) CatcherFunc

Implementation Notes

Configuration Lifecycle
  1. Create a Config with required fields.
  2. Call SetDefaults() to fill optional fields (done automatically by New).
  3. Call Valid() to verify the configuration (done automatically by New).
  4. Pass to New() to create a client.
  5. Configuration becomes immutable and bound to the client.
Connection Flow
  1. Connect() initiates the first connection attempt.
  2. On success, OnConnect callback is invoked.
  3. OnSession callback runs (blocks until session ends).
  4. On disconnection, OnDisconnect callback is invoked.
  5. WaitReconnect determines the retry delay.
  6. Process repeats until context cancellation or fatal error.
Session Handler Guidelines

The OnSession callback should:

  • Block until the session is complete.
  • Handle all protocol-specific logic.
  • Return nil or a non-fatal error to trigger reconnection.
  • Return ErrDoNotReconnect, possibly wrapped, to stop the client.
  • Respect context cancellation for a graceful wind-down; a session left blocked on a read is unblocked by shutdown closing the connection.

Integration with darvaza.org/x/net

The reconnect client integrates seamlessly with other darvaza.org/x/net components:

  • Uses the same net.Dialer interface.
  • Supports all standard network configurations.
  • Compatible with the bind package for advanced binding.

Example: Protocol Implementation

import (
    "context"
    "net"
    "time"

    "darvaza.org/x/net/reconnect"
)

func createClient(addr string) (*reconnect.Client, error) {
    cfg := &reconnect.Config{
        Remote:         addr,
        ReconnectDelay: 5 * time.Second,

        OnConnect: func(ctx context.Context, conn net.Conn) error {
            // Send initial handshake.
            return sendHandshake(conn)
        },

        OnSession: func(ctx context.Context) error {
            // Main protocol loop.
            for {
                select {
                case <-ctx.Done():
                    return ctx.Err()
                case msg := <-messages:
                    if err := processMessage(msg); err != nil {
                        return err
                    }
                }
            }
        },

        OnError: func(ctx context.Context, conn net.Conn, err error) error {
            if isTemporary(err) {
                // Retry on temporary errors.
                return nil
            }
            // Stop on permanent errors.
            return reconnect.ErrDoNotReconnect
        },
    }

    return reconnect.New(cfg)
}
Example: Unix Domain Socket Connection
import (
    "context"
    "net"
    "time"

    "darvaza.org/x/net/reconnect"
)

func createUnixClient() (*reconnect.Client, error) {
    cfg := &reconnect.Config{
        // Automatically detected as Unix socket
        Remote:         "/var/run/myapp.sock",
        ReconnectDelay: 5 * time.Second,

        OnConnect: func(ctx context.Context, conn net.Conn) error {
            // Unix socket connected
            logger.Info().Printf("Connected to Unix: %s", conn.RemoteAddr())
            return nil
        },

        OnSession: func(ctx context.Context) error {
            // Handle Unix socket communication
            return handleUnixProtocol(ctx)
        },
    }

    return reconnect.New(cfg)
}

// Alternative: explicit Unix socket prefix
func createExplicitUnixClient() (*reconnect.Client, error) {
    cfg := &reconnect.Config{
        Remote: "unix:/tmp/app.sock",
        // ... rest of configuration
    }
    return reconnect.New(cfg)
}

Network Type Auto-Detection

The client automatically determines the network type based on the Remote string:

Pattern Network Type Example
unix: prefix Unix socket unix:/var/run/app.sock
@ prefix Unix socket (abstract namespace, Linux) @app
Absolute path (/) Unix socket /tmp/socket
Ends with .sock Unix socket ./app.sock, run/app.sock
All others TCP example.com:8080, 192.168.1.1:443

Documentation

Overview

Package reconnect implements a generic retrying network client.

Index

Constants

View Source 
const (
	// LogFieldAddress is the field name used to store the address
	// when logging.
	LogFieldAddress = "addr"

	// LogFieldError is the field name used to store the error
	// when logging.
	LogFieldError = slog.ErrorFieldName
)
View Source 
const (
	// NetworkTCP represents TCP network type
	NetworkTCP = "tcp"
	// NetworkUnix represents Unix domain socket network type
	NetworkUnix = "unix"
	// MaxUNIXSocketPathLength is the maximum length for UNIX domain socket paths
	// Limited by sockaddr_un.sun_path (108 bytes including null terminator on Linux)
	MaxUNIXSocketPathLength = 107
)
View Source 
const (
	// DefaultWaitReconnect specifies how long [NewConstantWaiter]
	// waits between reconnection attempts by default.
	DefaultWaitReconnect = 5 * time.Second
)

Variables

View Source 
var (
	// ErrAbnormalConnect indicates the dialer didn't return error
	// nor connection.
	ErrAbnormalConnect = core.QuietWrap(syscall.ECONNABORTED, "abnormal response")

	// ErrDoNotReconnect indicates the Waiter
	// instructed us to not reconnect
	ErrDoNotReconnect = errors.New("don't reconnect")

	// ErrNotConnected indicates the [Client] isn't currently connected.
	// It wraps [ErrClosed] so a single errors.Is target covers both a
	// closed client and the not-connected window.
	ErrNotConnected = core.QuietWrap(ErrClosed, "client not connected")

	// ErrRunning indicates the [Client] has already been started.
	ErrRunning = core.QuietWrap(syscall.EBUSY, "client already running")

	// ErrClosed indicates the [Client] or [StreamSession] has
	// already been shut down. It wraps the workgroup's sentinel so
	// the shutdown signal still matches the one the group returns
	// across the lifecycle stack.
	ErrClosed = core.QuietWrap(errors.ErrClosed, "already closed")

	// ErrNameEmpty indicates a name is empty. It wraps [core.ErrInvalid]
	// so a caller matching the invalid-argument family with errors.Is
	// catches it.
	ErrNameEmpty = core.QuietWrap(core.ErrInvalid, "name missing")

	// ErrNameTooLong indicates a name exceeds maximum length. It wraps
	// [core.ErrInvalid] for the same reason.
	ErrNameTooLong = core.QuietWrap(core.ErrInvalid, "name too long")
)
View Source 
var (
	// ErrConfigBusy indicates the [Config] is in use and can't
	// be used to create another [Client].
	ErrConfigBusy = core.QuietWrap(fs.ErrPermission, "config already in use")
)

Functions

func IsClosed added in v0.8.1 

func IsClosed(err error) bool

IsClosed reports whether err indicates the Client has been shut down, matching ErrClosed anywhere in the chain. Because ErrNotConnected wraps ErrClosed, IsClosed is the broad companion to IsNotConnected: it is true for both a fully closed client and the transient not-connected window, whereas IsNotConnected matches only the latter.

func IsFatal 

func IsFatal(err error) bool

IsFatal tells if the error means the connection should be closed and not retried. Only ErrDoNotReconnect, possibly wrapped, is considered fatal; anything else is treated as recoverable.

IsFatal classifies connection errors seen inside the reconnect loop. Caller-misuse errors are reported at setup time and never reach that decision; they extend core.ErrInvalid, so match them with errors.Is against that sentinel instead.

func IsNonError 

func IsNonError(err error) bool

IsNonError reports whether the error represents a user-initiated shutdown instead of an actual failure.

func IsNotConnected added in v0.8.1 

func IsNotConnected(err error) bool

IsNotConnected reports whether err indicates the Client had no session when a request was attempted, matching ErrNotConnected anywhere in the chain. A fully shut-down client surfaces ErrClosed without ErrNotConnected wrapping it; match ErrClosed instead to cover both the closed and not-connected cases with a single target.

func NewConstantWaiter 

func NewConstantWaiter(d time.Duration) func(context.Context) error

NewConstantWaiter blocks for a given amount of time, or until the context is cancelled. If the given duration is zero, DefaultWaitReconnect is used. If negative, reconnecting is disabled, failing with ErrDoNotReconnect.

func NewDoNotReconnectWaiter 

func NewDoNotReconnectWaiter(err error) func(context.Context) error

NewDoNotReconnectWaiter returns a Waiter that stops reconnection attempts, failing with the given error, or ErrDoNotReconnect when nil. The context's error takes precedence if the context has already terminated.

func NewImmediateErrorWaiter 

func NewImmediateErrorWaiter(err error) func(context.Context) error

NewImmediateErrorWaiter returns a Waiter that doesn't wait. It returns the context's error if the context has already terminated, or the given error otherwise. A nil error allows an immediate reconnection attempt.

func ParseRemote added in v0.6.1 

func ParseRemote(remote string) (network, address string, err error)

ParseRemote determines the network type and address from a remote string. It supports: - "unix:/path/to/socket" - explicit Unix socket. - "/path/to/socket" - Unix socket (absolute path). - "@abstract-name" - abstract Unix socket. - "path/to/file.sock" - Unix socket (ends with .sock). - "host:port" - TCP socket.

func TimeoutToAbsoluteTime 

func TimeoutToAbsoluteTime(base time.Time, d time.Duration) time.Time

TimeoutToAbsoluteTime adds the given time.Duration to a base time.Time. If the duration is negative, a zero time.Time will be returned. If the base is zero, the current time will be used.

func ValidateRemote added in v0.6.1 

func ValidateRemote(remote string) error

ValidateRemote validates a remote address for use with reconnect clients. It returns nil if the address is valid for either TCP or Unix socket connection. For TCP addresses, it validates host:port format. For Unix socket addresses, it accepts the address as-is.

Types

type CatcherFunc added in v0.3.0 

type CatcherFunc func(context.Context, error) error

CatcherFunc is a catch function for WorkGroup.GoCatch.

func NewCatchFunc added in v0.3.0 

func NewCatchFunc(nonErrors ...error) CatcherFunc

NewCatchFunc creates a CatcherFunc turning any of the given errors into nil.

type Client 

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

Client is a reconnecting network client.

func Must 

func Must(cfg *Config, options ...OptionFunc) *Client

Must is like New but it panics on errors.

func New 

func New(cfg *Config, options ...OptionFunc) (*Client, error)

New creates a new Client using the given Config and options.

func (*Client) Close 

func (c *Client) Close() error

Close terminates the current connection, if any. The Client keeps running and will reconnect; use Client.Shutdown to stop it.

func (*Client) Config 

func (c *Client) Config() *Config

Config returns the Config object used when Client.Reload is called.

func (*Client) Connect 

func (c *Client) Connect() error

Connect launches the Client, failing with ErrRunning when called more than once. A nil return means the reconnection loop has started, not that a connection is established — a failed first dial is retried in the background like any other disconnection. Calling it on a Client that has already been shut down fails with ErrClosed.

func (*Client) Done 

func (c *Client) Done() <-chan struct{}

Done returns a channel that is closed once the Client workers have finished. Use Client.Err to learn the cancellation reason.

func (*Client) Err 

func (c *Client) Err() error

Err returns the cancellation reason. It will return nil if the cause was initiated by the user.

func (*Client) Go 

func (c *Client) Go(funcs ...WorkerFunc)

Go spawns a goroutine within the Client's context. Submissions after shutdown are no-ops: the worker is dropped rather than run with an already-cancelled context.

func (*Client) GoCatch added in v0.3.0 

func (c *Client) GoCatch(run WorkerFunc, catch CatcherFunc)

GoCatch spawns a goroutine within the Client's context, optionally allowing filtering the error to stop cascading. Submissions after shutdown are no-ops, as with Client.Go.

func (*Client) LocalAddr added in v0.2.4 

func (c *Client) LocalAddr() net.Addr

LocalAddr returns the local address if connected.

func (*Client) Read 

func (c *Client) Read(p []byte) (int, error)

Read reads from the current connection, if connected.

func (*Client) Reload 

func (c *Client) Reload() error

Reload attempts to apply changes done to the Config since the last time, or since created.

func (*Client) RemoteAddr added in v0.2.4 

func (c *Client) RemoteAddr() net.Addr

RemoteAddr returns the remote address if connected.

func (*Client) ResetDeadline 

func (c *Client) ResetDeadline() error

ResetDeadline sets the connection's read and write deadlines using the default values.

func (*Client) ResetReadDeadline 

func (c *Client) ResetReadDeadline() error

ResetReadDeadline resets the connection's read deadline using the default duration.

func (*Client) ResetWriteDeadline 

func (c *Client) ResetWriteDeadline() error

ResetWriteDeadline resets the connection's write deadline using the default duration.

func (*Client) SetDeadline 

func (c *Client) SetDeadline(read, write time.Duration) error

SetDeadline sets the connection's read and write deadlines. If write is zero but read is positive, write is set using the same value as read. Zero or negative can be used to disable the deadline.

func (*Client) SetReadDeadline added in v0.2.3 

func (c *Client) SetReadDeadline(d time.Duration) error

SetReadDeadline sets the connection's read deadline to the specified duration. Use zero or negative to disable it.

func (*Client) SetWriteDeadline added in v0.2.3 

func (c *Client) SetWriteDeadline(d time.Duration) error

SetWriteDeadline sets the connection's write deadline to the specified duration. Use zero or negative to disable it.

func (*Client) Shutdown 

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

Shutdown initiates a shutdown and waits until the workers are done, or the given context times out.

func (*Client) Wait 

func (c *Client) Wait() error

Wait blocks until the Client workers have finished, and returns the cancellation reason, nil if the shutdown was user-initiated.

func (*Client) WithDebug added in v0.2.4 

func (c *Client) WithDebug(addr net.Addr) (slog.Logger, bool)

WithDebug gets a logger at Debug level optionally annotated by an IP address. If the Debug log-level is disabled, it will return `nil, false`.

func (*Client) WithError 

func (c *Client) WithError(addr net.Addr, err error) (slog.Logger, bool)

WithError gets a logger at Error level optionally annotated by an IP address. If the Error log-level is disabled, it will return `nil, false`.

func (*Client) WithInfo 

func (c *Client) WithInfo(addr net.Addr) (slog.Logger, bool)

WithInfo gets a logger at Info level optionally annotated by an IP address. If the Info log-level is disabled, it will return `nil, false`.

func (*Client) Write 

func (c *Client) Write(p []byte) (int, error)

Write writes to the current connection, if connected.

type Config 

type Config struct {
	Context context.Context
	Logger  slog.Logger

	// WaitReconnect is a helper used to wait between re-connection attempts.
	WaitReconnect Waiter

	// OnSocket is called, when defined, against the raw socket before attempting to
	// connect
	OnSocket func(context.Context, syscall.RawConn) error
	// OnConnect is called, when defined, immediately after the connection is established
	// but before the session is created.
	OnConnect func(context.Context, net.Conn) error
	// OnSession, when defined, owns the connection and is expected
	// to block until the session is done. Returning nil or a
	// non-fatal error leads to a reconnection attempt; return
	// [ErrDoNotReconnect], possibly wrapped, to stop the [Client].
	OnSession func(context.Context) error
	// OnDisconnect is called after closing the connection and can be used to
	// prevent further connection retries by returning [ErrDoNotReconnect].
	OnDisconnect func(context.Context, net.Conn) error
	// OnError is called after all errors, and its return value replaces
	// the error for the reconnection logic. Return nil to discard the
	// error, or [ErrDoNotReconnect] to stop the [Client].
	OnError func(context.Context, net.Conn, error) error

	// Remote indicates the remote endpoint: TCP "host:port" or Unix socket path,
	// e.g., "/path/to/socket" or "unix:/path".
	Remote string

	// KeepAlive indicates the value to be set to TCP connections
	// for the low-level keep-alive messages.
	KeepAlive time.Duration `default:"5s"`
	// DialTimeout indicates how long are we willing to wait for new
	// connections getting established.
	DialTimeout time.Duration `default:"2s"`
	// ReadTimeout is the default read deadline for the connection,
	// applied via [Client.ResetReadDeadline] and [Client.ResetDeadline].
	// It is not set automatically on new connections.
	// Zero or negative disables the deadline.
	ReadTimeout time.Duration `default:"2s"`
	// WriteTimeout is the default write deadline for the connection,
	// applied via [Client.ResetWriteDeadline] and [Client.ResetDeadline].
	// It is not set automatically on new connections.
	// Zero or negative disables the deadline, except [Client.ResetDeadline]
	// substitutes ReadTimeout when WriteTimeout is zero (see [Client.SetDeadline]).
	WriteTimeout time.Duration `default:"2s"`
	// ReconnectDelay specifies how long to wait between re-connections
	// unless [WaitReconnect] is specified. Zero means
	// [DefaultWaitReconnect], and negative implies reconnecting
	// is disabled.
	ReconnectDelay time.Duration
	// contains filtered or unexported fields
}

Config describes the operation of the Client.

func (*Config) ExportDialer 

func (cfg *Config) ExportDialer() net.Dialer

ExportDialer creates a net.Dialer from the Config.

func (*Config) SetDefaults 

func (cfg *Config) SetDefaults() error

SetDefaults fills any gap in the config.

func (*Config) Valid 

func (cfg *Config) Valid() error

Valid checks if the Config is fit to be used.

type OptionFunc 

type OptionFunc func(*Config) error

An OptionFunc modifies a Config before Config.SetDefaults and Config.Valid run.

type Shutdowner added in v0.3.0 

type Shutdowner interface {
	Shutdown(context.Context) error
}

A Shutdowner is an object that provides a Shutdown method that takes a context with deadline to shut down all associated workers.

type StreamSession added in v0.2.2 

type StreamSession[Input, Output any] struct {

	// Conn specifies the underlying connection
	Conn io.ReadWriteCloser
	// Context is an optional [context.Context] to allow cascading cancellations.
	Context context.Context

	// Split identifies the next encoded [Input] type in the inbound stream.
	// If not set, [bufio.ScanLines] will be used.
	Split bufio.SplitFunc
	// Marshal is used, if MarshalTo isn't set, to encode an [Output] type.
	// If neither is set, [StreamSession.Spawn] will fail.
	Marshal func(Output) ([]byte, error)
	// MarshalTo, if set, is used to write the encoded representation of
	// an [Output] type.
	MarshalTo func(Output, io.Writer) error
	// Unmarshal is used to decode an [Input] type previously identified
	// by [StreamSession.Split].
	// If not set, [StreamSession.Spawn] will fail.
	Unmarshal func([]byte) (Input, error)

	// SetReadDeadline is an optional hook called before reading a message
	SetReadDeadline func() error
	// SetWriteDeadline is an optional hook called before writing a message
	SetWriteDeadline func() error
	// UnsetReadDeadline is an optional hook called after having read a message
	UnsetReadDeadline func() error
	// UnsetWriteDeadline is an optional hook called after having written a message
	UnsetWriteDeadline func() error

	// OnError is optionally called when an error occurs
	OnError func(error)

	// QueueSize specifies how many [Output] type entries can be buffered
	// for delivery before [StreamSession.Send] blocks.
	QueueSize uint
	// contains filtered or unexported fields
}

StreamSession provides an asynchronous stream session using message types for receiving and sending. Exported fields are configured before calling StreamSession.Spawn and must not be modified afterwards. The session must be spawned before using any other method.

func (*StreamSession[_, _]) Close added in v0.2.2 

func (s *StreamSession[_, _]) Close() error

Close initiates a shutdown of the session.

func (*StreamSession[_, _]) Done added in v0.2.2 

func (s *StreamSession[_, _]) Done() <-chan struct{}

Done returns a channel that will be closed when all workers are done.

func (*StreamSession[Input, Output]) Err added in v0.3.0 

func (s *StreamSession[Input, Output]) Err() error

Err returns the error that initiated a shutdown.

func (*StreamSession[_, _]) Go added in v0.2.4 

func (s *StreamSession[_, _]) Go(funcs ...WorkerFunc)

Go spawns a goroutine within the session's context.

func (*StreamSession[_, _]) GoCatch added in v0.3.0 

func (s *StreamSession[_, _]) GoCatch(run WorkerFunc, catch CatcherFunc)

GoCatch spawns a goroutine within the session's context, and allows a catcher function to filter returned errors.

func (*StreamSession[Input, _]) Next added in v0.2.2 

func (s *StreamSession[Input, _]) Next() (Input, bool)

Next blocks until a new message is received, returning false once the inbound stream has ended.

func (*StreamSession[Input, _]) Recv added in v0.2.2 

func (s *StreamSession[Input, _]) Recv() <-chan Input

Recv returns the channel where inbound messages are delivered. The channel is closed when the inbound stream ends.

func (*StreamSession[_, Output]) Send added in v0.2.2 

func (s *StreamSession[_, Output]) Send(m Output) error

Send queues a message for asynchronous delivery, blocking while the queue is full. It fails with ErrClosed once the session has been shut down.

func (*StreamSession[_, _]) Shutdown added in v0.3.0 

func (s *StreamSession[_, _]) Shutdown(ctx context.Context) error

Shutdown initiates a shutdown and waits until it's done or the given context has expired.

func (*StreamSession[_, _]) Spawn added in v0.2.2 

func (s *StreamSession[_, _]) Spawn() error

Spawn starts the StreamSession's workers. It fails if the session has already been started, or if Conn, Unmarshal, or a marshalling function is missing.

func (*StreamSession[_, _]) Wait added in v0.2.2 

func (s *StreamSession[_, _]) Wait() error

Wait blocks until all workers are done.

type Waiter 

type Waiter func(context.Context) error

A Waiter is a function that blocks between reconnection attempts. It returns nil when the Client is good to try again, or an error to stop reconnecting.

type WorkGroup added in v0.3.0 

type WorkGroup interface {
	Go(...WorkerFunc)
	GoCatch(WorkerFunc, CatcherFunc)

	Shutdown(context.Context) error

	Wait() error
	Done() <-chan struct{}
	Err() error
}

A WorkGroup is an error group interface. Submissions through Go and GoCatch after shutdown are no-ops.

type WorkerFunc added in v0.3.0 

type WorkerFunc func(context.Context) error

WorkerFunc is a run function for WorkGroup.GoCatch.

func NewShutdownFunc added in v0.3.0 

func NewShutdownFunc(s Shutdowner, tio time.Duration) WorkerFunc

NewShutdownFunc creates a shutdown WorkerFunc, optionally with a deadline.

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.