OddSockets Go SDK Documentation

Overview & Features

The OddSockets Go SDK provides a powerful, idiomatic Go interface for real-time messaging with full goroutine safety and excellent performance characteristics.

Idiomatic Go

Follows Go conventions with proper error handling, context support, and clean APIs.

Goroutine Safe

Thread-safe operations with proper synchronization for concurrent usage.

Type Safety

Strong typing with comprehensive struct definitions and interface contracts.

High Performance

Optimized for low latency with efficient WebSocket connections and minimal allocations.

Cost Effective

No per-message pricing, industry-standard 32KB message limits, transparent pricing.

Context Support

Full context.Context support for cancellation, timeouts, and request tracing.

Installation

bash

go get github.com/oddsocketsai/go-sdk

go

module your-app

go 1.19

require (
    github.com/oddsocketsai/go-sdk v1.0.0
)

bash

git clone https://github.com/oddsocketsai/go-sdk.git
cd go-sdk
go mod tidy

Quick Start

Basic Usage

go

package main

import (
    "context"
    "fmt"
    "log"
    
    "github.com/oddsocketsai/go-sdk/oddsockets"
)

func main() {
    // Create client
    client, err := oddsockets.NewClient(&oddsockets.Config{
        APIKey: "ak_live_1234567890abcdef",
    })
    if err != nil {
        log.Fatal(err)
    }
    defer client.Close()

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

    // Get channel
    channel := client.Channel("my-channel")

    // Subscribe to messages
    err = channel.Subscribe(ctx, func(msg *oddsockets.Message) {
        fmt.Printf("Received: %+v\n", msg)
    })
    if err != nil {
        log.Fatal(err)
    }

    // Publish a message
    err = channel.Publish(ctx, "Hello, World!")
    if err != nil {
        log.Fatal(err)
    }
}

With Error Handling

go

package main

import (
    "context"
    "fmt"
    "log"
    "time"
    
    "github.com/oddsocketsai/go-sdk/oddsockets"
)

func main() {
    // Create client with options
    client, err := oddsockets.NewClient(&oddsockets.Config{
        APIKey:            "ak_live_1234567890abcdef",
        UserID:            "user123",
        AutoConnect:       true,
        ReconnectAttempts: 5,
        HeartbeatInterval: 30 * time.Second,
    })
    if err != nil {
        log.Fatal("Failed to create client:", err)
    }
    defer client.Close()

    // Set up event handlers
    client.OnConnecting(func() {
        fmt.Println("🔄 Connecting...")
    })
    
    client.OnConnected(func() {
        fmt.Println("✅ Connected!")
    })
    
    client.OnDisconnected(func(reason string) {
        fmt.Printf("❌ Disconnected: %s\n", reason)
    })
    
    client.OnError(func(err error) {
        fmt.Printf("❌ Error: %v\n", err)
    })

    // Connect with timeout
    ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
    defer cancel()
    
    if err := client.Connect(ctx); err != nil {
        log.Fatal("Failed to connect:", err)
    }

    // Get channel and subscribe
    channel := client.Channel("my-channel")
    
    err = channel.Subscribe(ctx, func(msg *oddsockets.Message) {
        fmt.Printf("📨 Message: %s from %s\n", msg.Data, msg.UserID)
    }, &oddsockets.SubscribeOptions{
        EnablePresence: true,
        RetainHistory:  true,
        MaxHistory:     50,
    })
    if err != nil {
        log.Fatal("Failed to subscribe:", err)
    }

    // Publish with metadata
    err = channel.Publish(ctx, map[string]interface{}{
        "text": "Hello from Go SDK!",
        "user": "gopher",
    }, &oddsockets.PublishOptions{
        TTL:      3600,
        Metadata: map[string]interface{}{"priority": "high"},
    })
    if err != nil {
        log.Fatal("Failed to publish:", err)
    }

    // Keep the program running
    select {}
}

Concurrent Usage

go

package main

import (
    "context"
    "fmt"
    "log"
    "sync"
    "time"
    
    "github.com/oddsocketsai/go-sdk/oddsockets"
)

func main() {
    client, err := oddsockets.NewClient(&oddsockets.Config{
        APIKey: "ak_live_1234567890abcdef",
    })
    if err != nil {
        log.Fatal(err)
    }
    defer client.Close()

    ctx := context.Background()
    if err := client.Connect(ctx); err != nil {
        log.Fatal(err)
    }

    var wg sync.WaitGroup
    
    // Start multiple goroutines for concurrent operations
    for i := 0; i < 10; i++ {
        wg.Add(1)
        go func(id int) {
            defer wg.Done()
            
            channelName := fmt.Sprintf("channel-%d", id)
            channel := client.Channel(channelName)
            
            // Subscribe
            err := channel.Subscribe(ctx, func(msg *oddsockets.Message) {
                fmt.Printf("Channel %s received: %+v\n", channelName, msg)
            })
            if err != nil {
                log.Printf("Failed to subscribe to %s: %v", channelName, err)
                return
            }
            
            // Publish messages
            for j := 0; j < 5; j++ {
                message := fmt.Sprintf("Message %d from goroutine %d", j, id)
                if err := channel.Publish(ctx, message); err != nil {
                    log.Printf("Failed to publish to %s: %v", channelName, err)
                }
                time.Sleep(100 * time.Millisecond)
            }
        }(i)
    }
    
    wg.Wait()
}

Configuration

Client Configuration

go

client, err := oddsockets.NewClient(&oddsockets.Config{
    APIKey:            "your-api-key",           // Required: Your OddSockets API key
    UserID:            "user-id",                // Optional: User identifier
    AutoConnect:       true,                     // Optional: Auto-connect on creation
    ReconnectAttempts: 5,                        // Optional: Max reconnection attempts
    HeartbeatInterval: 30 * time.Second,         // Optional: Heartbeat interval
    ConnectTimeout:    15 * time.Second,         // Optional: Connection timeout
    Logger:            log.New(os.Stdout, "", 0), // Optional: Custom logger
})

Channel Options

go

// Subscribe with options
err = channel.Subscribe(ctx, messageHandler, &oddsockets.SubscribeOptions{
    EnablePresence: true,                        // Enable presence tracking
    RetainHistory:  true,                        // Retain message history
    MaxHistory:     100,                         // Maximum history size
    Filter:         "user.premium == true",      // Message filter expression
})

// Publish with options
err = channel.Publish(ctx, message, &oddsockets.PublishOptions{
    TTL:           3600,                         // Time to live (seconds)
    Metadata:      map[string]interface{}{       // Additional metadata
        "priority": "high",
        "source":   "go-sdk",
    },
    StoreInHistory: true,                        // Store in message history
})

Context Usage

go

// With timeout
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()

// With cancellation
ctx, cancel := context.WithCancel(context.Background())
defer cancel()

// With deadline
deadline := time.Now().Add(5 * time.Minute)
ctx, cancel := context.WithDeadline(context.Background(), deadline)
defer cancel()

// With values for tracing
ctx = context.WithValue(ctx, "requestID", "req-123")
ctx = context.WithValue(ctx, "userID", "user-456")

Examples

Explore comprehensive examples demonstrating the OddSockets Go SDK in action:

View All Examples

Browse all available examples including basic usage, concurrent operations, and production patterns

Enhanced Features

Beyond core pub/sub, OddSockets ships a Slack-like enhanced surface — reactions, typing indicators, threads, read receipts, presence/status, notifications, DMs, channel management, message editing and search. It lives on client.Enhanced. The pattern is always the same: send an action with a client.Enhanced.* method, then receive the paired broadcast with client.On("<event>", handler).

Typing & Reactions

go

// Receive-path: broadcasts from other users on the channel
client.On("user_typing", func(_ oddsockets.EventType, data interface{}) {
    fmt.Println("user is typing")
})
client.On("reaction_added", func(_ oddsockets.EventType, data interface{}) {
    fmt.Println("reaction added")
})

// Send-path: enhanced actions over the live socket
client.Enhanced.StartTyping("alice", "room-42")

client.Enhanced.AddReaction(oddsockets.ReactionParams{
    MessageID: "msg-1",
    Channel:   "room-42",
    Emoji:     ":thumbsup:",
    UserID:    "alice",
    UserName:  "Alice",
})

Threads

go

client.On("thread_reply", func(_ oddsockets.EventType, data interface{}) {
    fmt.Println("new thread reply")
})

result, err := client.Enhanced.ThreadReply(oddsockets.ThreadReplyParams{
    Channel:         "room-42",
    ParentMessageID: "msg-1",
    Message:         "Replying in the thread",
    UserID:          "alice",
    UserName:        "Alice",
})
if err != nil {
    log.Fatal(err)
}
fmt.Printf("thread reply: %v\n", result)

Enhanced surface

Each area exposes methods on client.Enhanced; the worker broadcasts the paired events which you handle with client.On(...). Query methods (Get*, Search*) return (map[string]interface{}, error).

Typing — StartTyping, StopTyping → user_typing, user_stopped_typing
Reactions — AddReaction, RemoveReaction, GetReactions → reaction_added, reaction_removed
Threads — ThreadReply, GetThread, SubscribeThread, FollowThread, MarkThreadRead → thread_reply, thread_subscribed, thread_followed, thread_read_updated
Read receipts — MarkRead, MarkAllRead, GetUnreadCounts → user_read, unread_count_updated, all_marked_read
Messages — EditMessage, DeleteMessage, PinMessage, UnpinMessage, GetPinnedMessages, SearchMessages → message_edited, message_deleted, message_pinned, message_unpinned
Presence & status — SetStatus, SetCustomStatus, SetDND, GetUserPresence → user_status_changed, custom_status_updated, dnd_status_changed
Channels — CreateChannel, UpdateChannel, ArchiveChannel, InviteToChannel, JoinChannel, LeaveChannel → channel_created, channel_updated, user_invited, user_joined_channel, user_left_channel
DMs — CreateDM, SendDM, GetDMConversations → dm_created, dm_received
Notifications — SubscribeNotifications, GetNotifications, MarkNotificationRead, ClearNotifications → notification, notification_read, notifications_cleared
Search — SearchMessages, SearchInChannel, SearchByUser, FilterMessages → (return values)

For any worker event not wrapped above, subscribe with the raw client.On("<event>", handler) API — all enhanced broadcasts are forwarded onto the client surface.

Challenges & Leaderboards

Challenges, leaderboards and achievements build on the same live socket. The send side lives on the enhanced surface (client.Enhanced); request/query methods resolve with the worker's reply as (map[string]interface{}, error), while progress and achievement calls are fire-and-forget. Inbound broadcasts arrive on the client event surface — subscribe with the client's normal client.On("<event>", handler).

go

// Create a ranked leaderboard, report progress, read standings, then finalize
client.Enhanced.CreateChallenge(oddsockets.CreateChallengeParams{
    ChallengeID: "weekly-sprint",
    Metric:      "points",
    Ranked:      true,
})
client.Enhanced.ReportProgress(oddsockets.ReportProgressParams{
    ChallengeID: "weekly-sprint",
    Value:       120,
})
standings, err := client.Enhanced.GetStandings(oddsockets.GetStandingsParams{
    ChallengeID: "weekly-sprint",
    Limit:       10,
})
if err != nil {
    log.Fatal(err)
}
fmt.Printf("standings: %v\n", standings)
client.Enhanced.CompleteChallenge(oddsockets.CompleteChallengeParams{
    ChallengeID: "weekly-sprint",
    Outcome:     "completed",
})

Send-side methods

Request/query methods (GetStandings, GetAchievements, CreateChallenge, CompleteChallenge, the invite calls) return (map[string]interface{}, error) and resolve with the worker's ack. ReportProgress and UnlockAchievement are fire-and-forget.

CreateChallenge — create a challenge/leaderboard. Ack challenge_create_success.
ReportProgress — fire-and-forget metric progress. No ack.
CompleteChallenge — finalize with an outcome. Ack challenge_complete_success.
UnlockAchievement — fire-and-forget; pass PercentComplete (0–100). No ack.
GetStandings — request top-N + caller rank. Ack challenge_standings_success.
GetAchievements — query achievement state. Ack achievement_state.
SendChallengeInvite — directed invite to another user. Ack challenge_invite_success.
ReplyChallengeInvite — accept/decline an invite. Ack challenge_reply_success.
CancelChallengeInvite — cancel a sent invite. Ack challenge_invite_cancel_success.
GetChallengeInvites — list pending invites. Ack challenge_invites.

Completion outcomes

CompleteChallenge takes an Outcome from this vocabulary:

completed — win (rank 1)
failed — loss
tied — draw
conceded — resign / concede
expired — timed out

Progressive achievements

UnlockAchievement always emits the wire event achievement_unlock; the worker is authoritative and derives the outbound broadcast from PercentComplete. A value < 100 broadcasts achievement_progress (status in_progress); a value >= 100 or omitted broadcasts achievement_unlock (status unlocked). Always call UnlockAchievement — do not emit achievement_progress yourself.

Inbound broadcasts

Subscribe to these on the client surface with client.On("<event>", handler).

Room broadcasts — challenge_progress, leaderboard_rank_change, challenge_complete, achievement_unlock, achievement_progress
Directed (per-user) — challenge_invited, challenge_reply_received, challenge_invite_cancelled

Usage Analytics

Pull your tenant's headline usage tiles — monthly active users, daily active users, total messages published, and error rate — straight from the SDK, without hand-rolling a REST call. GetUsageStats() resolves the same four tiles the developer dashboard renders.

stats, err := client.GetUsageStats(context.Background())
if err != nil {
    log.Fatal(err)
}

// Each tile is a *int64 / *float64 — nil means "no data", so render an em-dash.
tile := func(v *int64) string {
    if v == nil {
        return "\u2014"
    }
    return fmt.Sprintf("%d", *v)
}
fmt.Println("MAU:     ", tile(stats.MAU))
fmt.Println("DAU:     ", tile(stats.DAU))
fmt.Println("Messages:", tile(stats.TotalMessages))

The four tiles

MAU — monthly active users for your owner scope
DAU — daily active users
TotalMessages — total messages published
ErrorRate — publish error rate, 0–1

Honesty rule — nil, never a fake zero

Each tile is a pointer (*int64 / *float64) that is a number or nil. A nil tile means that leg of the analytics pipeline is not live yet for your tenant — the SDK carries it through verbatim and never coerces it to 0. Render an em-dash (—) for a nil tile so you never show a fabricated zero.

Requires an API key

GetUsageStats() reads your owner-scoped analytics, so it needs an APIKey. Keyless / token-only clients have no owner scope to query and will return the error GetUsageStats requires an apiKey.

Performance & Compatibility

OddSockets Go SDK delivers excellent performance with broad Go version compatibility:

<50ms

Latency

99.9%

Uptime

32KB

Max Message

1M+

Messages/sec

Go Version Support

Go 1.19+ (recommended)
Go 1.18+ (supported)
Go Modules required
CGO not required

Platform Support

Linux (amd64, arm64)
macOS (amd64, arm64)
Windows (amd64)
Docker containers

Framework Integrations

The OddSockets Go SDK works seamlessly with popular Go frameworks and libraries. Here are examples showing integration patterns:

Gin Web Framework

go

package main

import (
    "context"
    "net/http"
    
    "github.com/gin-gonic/gin"
    "github.com/oddsocketsai/go-sdk/oddsockets"
)

type ChatService struct {
    client *oddsockets.Client
}

func NewChatService() (*ChatService, error) {
    client, err := oddsockets.NewClient(&oddsockets.Config{
        APIKey: "ak_live_1234567890abcdef",
    })
    if err != nil {
        return nil, err
    }
    
    if err := client.Connect(context.Background()); err != nil {
        return nil, err
    }
    
    return &ChatService{client: client}, nil
}

func (cs *ChatService) SendMessage(c *gin.Context) {
    var req struct {
        Channel string      `json:"channel" binding:"required"`
        Message interface{} `json:"message" binding:"required"`
    }
    
    if err := c.ShouldBindJSON(&req); err != nil {
        c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
        return
    }
    
    channel := cs.client.Channel(req.Channel)
    if err := channel.Publish(c.Request.Context(), req.Message); err != nil {
        c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
        return
    }
    
    c.JSON(http.StatusOK, gin.H{"status": "sent"})
}

func main() {
    chatService, err := NewChatService()
    if err != nil {
        panic(err)
    }
    
    r := gin.Default()
    r.POST("/send", chatService.SendMessage)
    r.Run(":8080")
}

Echo Framework

go

package main

import (
    "context"
    "net/http"
    
    "github.com/labstack/echo/v4"
    "github.com/labstack/echo/v4/middleware"
    "github.com/oddsocketsai/go-sdk/oddsockets"
)

type Server struct {
    echo   *echo.Echo
    client *oddsockets.Client
}

func NewServer() (*Server, error) {
    client, err := oddsockets.NewClient(&oddsockets.Config{
        APIKey: "ak_live_1234567890abcdef",
    })
    if err != nil {
        return nil, err
    }
    
    if err := client.Connect(context.Background()); err != nil {
        return nil, err
    }
    
    e := echo.New()
    e.Use(middleware.Logger())
    e.Use(middleware.Recover())
    
    server := &Server{
        echo:   e,
        client: client,
    }
    
    server.setupRoutes()
    return server, nil
}

func (s *Server) setupRoutes() {
    s.echo.POST("/channels/:channel/messages", s.publishMessage)
    s.echo.GET("/channels/:channel/history", s.getHistory)
    s.echo.GET("/channels/:channel/presence", s.getPresence)
}

func (s *Server) publishMessage(c echo.Context) error {
    channel := c.Param("channel")
    
    var message interface{}
    if err := c.Bind(&message); err != nil {
        return echo.NewHTTPError(http.StatusBadRequest, err.Error())
    }
    
    ch := s.client.Channel(channel)
    if err := ch.Publish(c.Request().Context(), message); err != nil {
        return echo.NewHTTPError(http.StatusInternalServerError, err.Error())
    }
    
    return c.JSON(http.StatusOK, map[string]string{"status": "published"})
}

func (s *Server) getHistory(c echo.Context) error {
    channel := c.Param("channel")
    
    ch := s.client.Channel(channel)
    history, err := ch.GetHistory(c.Request().Context(), &oddsockets.HistoryOptions{
        Count: 50,
    })
    if err != nil {
        return echo.NewHTTPError(http.StatusInternalServerError, err.Error())
    }
    
    return c.JSON(http.StatusOK, history)
}

func (s *Server) getPresence(c echo.Context) error {
    channel := c.Param("channel")
    
    ch := s.client.Channel(channel)
    presence, err := ch.GetPresence(c.Request().Context())
    if err != nil {
        return echo.NewHTTPError(http.StatusInternalServerError, err.Error())
    }
    
    return c.JSON(http.StatusOK, presence)
}

func main() {
    server, err := NewServer()
    if err != nil {
        panic(err)
    }
    
    server.echo.Logger.Fatal(server.echo.Start(":8080"))
}

Fiber Framework

go

package main

import (
    "context"
    "log"
    
    "github.com/gofiber/fiber/v2"
    "github.com/gofiber/fiber/v2/middleware/cors"
    "github.com/gofiber/fiber/v2/middleware/logger"
    "github.com/oddsocketsai/go-sdk/oddsockets"
)

func main() {
    // Initialize OddSockets client
    client, err := oddsockets.NewClient(&oddsockets.Config{
        APIKey: "ak_live_1234567890abcdef",
    })
    if err != nil {
        log.Fatal(err)
    }
    defer client.Close()
    
    if err := client.Connect(context.Background()); err != nil {
        log.Fatal(err)
    }
    
    // Initialize Fiber app
    app := fiber.New(fiber.Config{
        ErrorHandler: func(c *fiber.Ctx, err error) error {
            return c.Status(fiber.StatusInternalServerError).JSON(fiber.Map{
                "error": err.Error(),
            })
        },
    })
    
    // Middleware
    app.Use(logger.New())
    app.Use(cors.New())
    
    // Routes
    app.Post("/channels/:channel/publish", func(c *fiber.Ctx) error {
        channel := c.Params("channel")
        
        var body map[string]interface{}
        if err := c.BodyParser(&body); err != nil {
            return err
        }
        
        ch := client.Channel(channel)
        if err := ch.Publish(context.Background(), body); err != nil {
            return err
        }
        
        return c.JSON(fiber.Map{"status": "published"})
    })
    
    app.Get("/channels/:channel/subscribe", func(c *fiber.Ctx) error {
        channel := c.Params("channel")
        
        ch := client.Channel(channel)
        
        // Set up SSE headers
        c.Set("Content-Type", "text/event-stream")
        c.Set("Cache-Control", "no-cache")
        c.Set("Connection", "keep-alive")
        
        // Subscribe and stream messages
        err := ch.Subscribe(context.Background(), func(msg *oddsockets.Message) {
            c.WriteString("data: " + string(msg.Data) + "\n\n")
        })
        
        return err
    })
    
    log.Fatal(app.Listen(":8080"))
}

gRPC Integration

go

package main

import (
    "context"
    "log"
    "net"
    
    "google.golang.org/grpc"
    "github.com/oddsocketsai/go-sdk/oddsockets"
)

type ChatServer struct {
    client *oddsockets.Client
    // UnimplementedChatServiceServer
}

func NewChatServer() (*ChatServer, error) {
    client, err := oddsockets.NewClient(&oddsockets.Config{
        APIKey: "ak_live_1234567890abcdef",
    })
    if err != nil {
        return nil, err
    }
    
    if err := client.Connect(context.Background()); err != nil {
        return nil, err
    }
    
    return &ChatServer{client: client}, nil
}

func (s *ChatServer) SendMessage(ctx context.Context, req *SendMessageRequest) (*SendMessageResponse, error) {
    channel := s.client.Channel(req.Channel)
    
    err := channel.Publish(ctx, map[string]interface{}{
        "text":   req.Message,
        "userId": req.UserId,
    })
    if err != nil {
        return nil, err
    }
    
    return &SendMessageResponse{Success: true}, nil
}

func (s *ChatServer) Subscribe(req *SubscribeRequest, stream ChatService_SubscribeServer) error {
    channel := s.client.Channel(req.Channel)
    
    return channel.Subscribe(stream.Context(), func(msg *oddsockets.Message) {
        response := &MessageEvent{
            Channel:   req.Channel,
            Message:   string(msg.Data),
            UserId:    msg.UserID,
            Timestamp: msg.Timestamp.Unix(),
        }
        
        if err := stream.Send(response); err != nil {
            log.Printf("Failed to send message: %v", err)
        }
    })
}

func main() {
    server, err := NewChatServer()
    if err != nil {
        log.Fatal(err)
    }
    
    lis, err := net.Listen("tcp", ":50051")
    if err != nil {
        log.Fatal(err)
    }
    
    s := grpc.NewServer()
    RegisterChatServiceServer(s, server)
    
    log.Println("gRPC server listening on :50051")
    log.Fatal(s.Serve(lis))
}

Worker Pool Pattern

go

package main

import (
    "context"
    "fmt"
    "log"
    "sync"
    "time"
    
    "github.com/oddsocketsai/go-sdk/oddsockets"
)

type MessageProcessor struct {
    client   *oddsockets.Client
    workers  int
    jobQueue chan *oddsockets.Message
    wg       sync.WaitGroup
}

func NewMessageProcessor(workers int) (*MessageProcessor, error) {
    client, err := oddsockets.NewClient(&oddsockets.Config{
        APIKey: "ak_live_1234567890abcdef",
    })
    if err != nil {
        return nil, err
    }
    
    if err := client.Connect(context.Background()); err != nil {
        return nil, err
    }
    
    return &MessageProcessor{
        client:   client,
        workers:  workers,
        jobQueue: make(chan *oddsockets.Message, 100),
    }, nil
}

func (mp *MessageProcessor) Start(ctx context.Context) {
    // Start worker goroutines
    for i := 0; i < mp.workers; i++ {
        mp.wg.Add(1)
        go mp.worker(ctx, i)
    }
    
    // Subscribe to incoming messages
    channel := mp.client.Channel("work-queue")
    err := channel.Subscribe(ctx, func(msg *oddsockets.Message) {
        select {
        case mp.jobQueue <- msg:
        case <-ctx.Done():
            return
        default:
            log.Println("Job queue full, dropping message")
        }
    })
    if err != nil {
        log.Fatal("Failed to subscribe:", err)
    }
}

func (mp *MessageProcessor) worker(ctx context.Context, id int) {
    defer mp.wg.Done()
    
    for {
        select {
        case msg := <-mp.jobQueue:
            mp.processMessage(ctx, id, msg)
        case <-ctx.Done():
            return
        }
    }
}

func (mp *MessageProcessor) processMessage(ctx context.Context, workerID int, msg *oddsockets.Message) {
    fmt.Printf("Worker %d processing message: %s\n", workerID, msg.Data)
    
    // Simulate work
    time.Sleep(100 * time.Millisecond)
    
    // Send result to results channel
    resultChannel := mp.client.Channel("results")
    result := map[string]interface{}{
        "originalMessage": string(msg.Data),
        "processedBy":     workerID,
        "processedAt":     time.Now().Unix(),
    }
    
    if err := resultChannel.Publish(ctx, result); err != nil {
        log.Printf("Worker %d failed to publish result: %v", workerID, err)
    }
}

func (mp *MessageProcessor) Stop() {
    close(mp.jobQueue)
    mp.wg.Wait()
    mp.client.Close()
}

func main() {
    processor, err := NewMessageProcessor(5)
    if err != nil {
        log.Fatal(err)
    }
    defer processor.Stop()
    
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()
    
    processor.Start(ctx)
    
    // Keep running
    select {}
}