qpub

package module
v0.2.0 Latest Latest
Warning

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

 Go to latest Published: Sep 25, 2026 License: Apache-2.0 Imports: 13 Imported by: 0

README

QPub Go SDK

Official Go client for QPub real-time messaging (Channels and Queues).

Stability: v0.x — the public API may change before v1.0.0. Pin a release, for example:

go get github.com/qpubio/qpub-go@v0.2.0

See CHANGELOG.md for release notes.

Install (latest pseudo-version)

go get github.com/qpubio/qpub-go

Quick start

Socket (subscribe / publish)
package main

import (
    "context"
    "fmt"
    "log"

    "github.com/qpubio/qpub-go"
    "github.com/qpubio/qpub-go/channel"
)

func main() {
    socket := qpub.NewSocket(qpub.WithAPIKey("YOUR_PUBLIC_ID:YOUR_SECRET"))
    defer socket.Reset()

    ch := socket.Channels.Get("my-channel")
    err := ch.Subscribe(context.Background(), func(m qpub.Message) {
        fmt.Println("received", string(m.Data))
    }, channel.SubscribeOptions{})
    if err != nil {
        log.Fatal(err)
    }

    _ = ch.Publish(context.Background(), "Hello!", channel.PublishOptions{})
}
REST (publish / queues)
import (
    "context"
    "github.com/qpubio/qpub-go"
    "github.com/qpubio/qpub-go/channel"
)

rest := qpub.NewRest(qpub.WithAPIKey("YOUR_PUBLIC_ID:YOUR_SECRET"))
defer rest.Reset()

ch := rest.Channels.Get("my-channel")
_, err := ch.Publish(context.Background(), "Hello!", channel.PublishOptions{})

See examples/ and the docs/ folder:

License

Apache-2.0

Documentation

Overview

Package qpub is the official Go client for QPub real-time messaging.

Use NewSocket for WebSocket pub/sub and NewRest for HTTP channels and queues.

Index

Constants

This section is empty.

Variables

View Source 
var (
	WithAPIKey      = option.WithAPIKey
	WithAutoConnect = option.WithAutoConnect
	DefaultOption   = option.DefaultOption
)

WithAPIKey and WithAutoConnect are option helpers.

View Source 
var AuthEvents = struct {
	TokenUpdated string
	TokenExpired string
	TokenError   string
	AuthError    string
}{
	TokenUpdated: events.AuthTokenUpdated,
	TokenExpired: events.AuthTokenExpired,
	TokenError:   events.AuthTokenError,
	AuthError:    events.AuthError,
}

AuthEvents names for authentication callbacks.

View Source 
var ChannelEvents = struct {
	Initialized   string
	Subscribing   string
	Subscribed    string
	Unsubscribing string
	Unsubscribed  string
	Paused        string
	Resumed       string
	Failed        string
}{
	Initialized:   events.ChannelInitialized,
	Subscribing:   events.ChannelSubscribing,
	Subscribed:    events.ChannelSubscribed,
	Unsubscribing: events.ChannelUnsubscribing,
	Unsubscribed:  events.ChannelUnsubscribed,
	Paused:        events.ChannelPaused,
	Resumed:       events.ChannelResumed,
	Failed:        events.ChannelFailed,
}

ChannelEvents names for socket channel lifecycle callbacks.

View Source 
var ConnectionEvents = struct {
	Initialized  string
	Connecting   string
	Opened       string
	Connected    string
	Disconnected string
	Closing      string
	Closed       string
	Failed       string
}{
	Initialized:  events.ConnectionInitialized,
	Connecting:   events.ConnectionConnecting,
	Opened:       events.ConnectionOpened,
	Connected:    events.ConnectionConnected,
	Disconnected: events.ConnectionDisconnected,
	Closing:      events.ConnectionClosing,
	Closed:       events.ConnectionClosed,
	Failed:       events.ConnectionFailed,
}

ConnectionEvents names for connection lifecycle callbacks.

Functions

This section is empty.

Types

type AuthManager 

type AuthManager interface {
	Authenticate(ctx context.Context) (*option.AuthResponse, error)
	IsAuthenticated() bool
	ShouldAutoAuthenticate() bool
	GetAuthenticateURL(baseURL string) (string, error)
	RequestToken(ctx context.Context, request option.TokenRequest) (*option.AuthResponse, error)
	GetCurrentToken() string
	GetAuthHeaders() (map[string]string, error)
	GetToken() string
	ClearToken()
	GetAuthQueryParams() (string, error)
	Reset()
	GenerateToken(ctx context.Context, opts option.TokenOptions) (string, error)
	IssueToken(ctx context.Context, opts option.TokenOptions) (string, error)
	CreateTokenRequest(ctx context.Context, opts option.TokenOptions) (option.TokenRequest, error)
	On(event string, fn func(any))
}

AuthManager handles authentication.

type AuthResponse 

type AuthResponse = option.AuthResponse

type Connection 

type Connection interface {
	Connect(ctx context.Context) error
	Disconnect()
	IsConnected() bool
	WaitUntilConnected(ctx context.Context) error
	Ping(ctx context.Context) (time.Duration, error)
	Reset()
	IsResetting() bool
	On(event string, fn func(any))
}

Connection manages WebSocket connectivity.

type EnqueueOptions 

type EnqueueOptions = protocol.EnqueueOptions

type EnqueueResult 

type EnqueueResult = protocol.EnqueueResult

type Message 

type Message = protocol.Message

type Option 

type Option = option.Option

type OptionFunc 

type OptionFunc = option.OptionFunc

type OptionManager 

type OptionManager interface {
	Get() option.Option
	Set(partial option.Option)
	Reset()
}

OptionManager manages SDK configuration.

type Permission 

type Permission = option.Permission

type QueueJob 

type QueueJob = protocol.QueueJob

type Rest 

type Rest struct {
	OptionManager *option.Manager
	Auth          *auth.Manager
	Channels      *channel.RestManager
	Queues        *queue.Manager
	// contains filtered or unexported fields
}

Rest is the HTTP client for channels and queues.

func NewRest 

func NewRest(funcs ...option.OptionFunc) *Rest

NewRest creates a REST client.

func (*Rest) GetInstanceID 

func (r *Rest) GetInstanceID() string

GetInstanceID returns instance identifier.

func (*Rest) Reset 

func (r *Rest) Reset()

Reset clears client state.

type RestChannel 

type RestChannel interface {
	Name() string
	Publish(ctx context.Context, data interface{}, opts channel.PublishOptions) ([]byte, error)
	Reset()
}

RestChannel publishes over HTTP.

type RestQueueManager 

type RestQueueManager interface {
	Enqueue(ctx context.Context, queueName string, payload interface{}, opts protocol.EnqueueOptions) (protocol.EnqueueResult, error)
	GetJob(ctx context.Context, queueName, jobID string) (protocol.QueueJob, error)
	ListJobs(ctx context.Context, queueName string, opts protocol.ListJobsOptions) ([]protocol.QueueJob, error)
	CancelJob(ctx context.Context, queueName, jobID string) error
	RetryJob(ctx context.Context, queueName, jobID string) error
	GetConfig(ctx context.Context, queueName string) (protocol.QueueConfig, error)
	UpdateConfig(ctx context.Context, queueName string, opts protocol.UpdateQueueConfigOptions) (protocol.QueueConfig, error)
	RegisterWorker(ctx context.Context, opts protocol.RegisterWorkerOptions) (protocol.WorkerRegistration, error)
	Heartbeat(ctx context.Context, workerID string) (protocol.WorkerRegistration, error)
	Pull(ctx context.Context, queueName string, opts protocol.PullJobsOptions) ([]protocol.QueueJob, error)
	Ack(ctx context.Context, queueName, jobID string, opts protocol.AckJobOptions) error
	Nack(ctx context.Context, queueName, jobID string, opts protocol.NackJobOptions) error
	RunWorker(ctx context.Context, queueName string, handler func(context.Context, protocol.QueueJob) (interface{}, error), opts protocol.RunWorkerOptions) error
	StopWorker()
	Reset()
}

RestQueueManager manages queue jobs over REST.

type Socket 

type Socket struct {
	OptionManager *option.Manager
	Auth          *auth.Manager
	Connection    *connection.Conn
	Channels      *channel.SocketManager
	// contains filtered or unexported fields
}

Socket is the WebSocket client for realtime pub/sub.

func NewSocket 

func NewSocket(funcs ...option.OptionFunc) *Socket

NewSocket creates a Socket client.

func (*Socket) GetInstanceID 

func (s *Socket) GetInstanceID() string

GetInstanceID returns instance identifier.

func (*Socket) Reset 

func (s *Socket) Reset()

Reset tears down connection, channels, auth, and options (connection → channels → auth → options).

type SocketChannel 

type SocketChannel interface {
	Name() string
	Publish(ctx context.Context, data interface{}, opts channel.PublishOptions) error
	Subscribe(ctx context.Context, handler channel.MessageHandler, opts channel.SubscribeOptions) error
	Unsubscribe(ctx context.Context, opts ...channel.UnsubscribeOptions) error
	On(event string, fn func(any))
	Pause(bufferMessages bool)
	Resume()
	IsPaused() bool
	ClearBufferedMessages()
	Reset()
}

SocketChannel is real-time pub/sub.

type TokenOptions 

type TokenOptions = option.TokenOptions

type TokenRequest 

type TokenRequest = option.TokenRequest

Source Files

View all Source files

Directories

Path Synopsis
examples
basic command
Basic REST publish example (set QPUB_API_KEY).
 Basic REST publish example (set QPUB_API_KEY).
queue-worker command
Queue worker example (requires QPUB_API_KEY).
 Queue worker example (requires QPUB_API_KEY).
socket command
Socket subscribe example (requires QPUB_API_KEY).
 Socket subscribe example (requires QPUB_API_KEY).
token-auth command
Token auth: server-side CreateTokenRequest, client-side RequestToken.
 Token auth: server-side CreateTokenRequest, client-side RequestToken.
internal
apikey
crypto
emitter
jwt
logger
port
Package testing provides mocks and helpers for unit tests against the public API.
 Package testing provides mocks and helpers for unit tests against the public API.
transport
httpclient
ws

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.