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/jyswee/oddsockets-go-sdk

go

module your-app

go 1.19

require (
    github.com/jyswee/oddsockets-go-sdk v0.0.0-20260825091527-73b6c34ac5c2
)

bash

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

Quick Start

Basic Usage

go

package main

import (
    "context"
    "fmt"
    "log"
    
    "github.com/jyswee/oddsockets-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 — they are delivered on a channel you own
    messages := make(chan *oddsockets.Message, 100)
    if err := channel.Subscribe(ctx, messages, nil); err != nil {
        log.Fatal(err)
    }

    go func() {
        for msg := range messages {
            fmt.Printf("Received: %+v\n", msg.Data)
        }
    }()

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

With Error Handling

go

package main

import (
    "context"
    "fmt"
    "log"
    "time"
    
    "github.com/jyswee/oddsockets-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 — one registration method, keyed by event type
    client.On(oddsockets.EventConnected, func(evt oddsockets.EventType, data interface{}) {
        fmt.Println("✅ Connected!")
    })

    client.On(oddsockets.EventDisconnected, func(evt oddsockets.EventType, data interface{}) {
        fmt.Printf("❌ Disconnected: %v\n", data)
    })

    client.On(oddsockets.EventReconnected, func(evt oddsockets.EventType, data interface{}) {
        fmt.Println("🔄 Reconnected")
    })

    client.On(oddsockets.EventError, func(evt oddsockets.EventType, data interface{}) {
        fmt.Printf("❌ Error: %v\n", data)
    })

    // 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")
    
    messages := make(chan *oddsockets.Message, 100)
    err = channel.Subscribe(ctx, messages, &oddsockets.SubscribeOptions{
        EnablePresence: true,
        RetainHistory:  true,
    })
    if err != nil {
        log.Fatal("Failed to subscribe:", err)
    }

    go func() {
        for msg := range messages {
            fmt.Printf("📨 Message: %v from %s\n", msg.Data, msg.UserID)
        }
    }()

    // 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/jyswee/oddsockets-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
            messages := make(chan *oddsockets.Message, 32)
            if err := channel.Subscribe(ctx, messages, nil); err != nil {
                log.Printf("Failed to subscribe to %s: %v", channelName, err)
                return
            }

            go func() {
                for msg := range messages {
                    fmt.Printf("Channel %s received: %+v\n", channelName, msg.Data)
                }
            }()

            // Publish messages
            for j := 0; j < 5; j++ {
                message := fmt.Sprintf("Message %d from goroutine %d", j, id)
                if _, err := channel.Publish(ctx, message, nil); 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

Platform limits and plan ceilings, plus the versions and platforms this SDK supports:

32KB

Max message size

100M

Messages/mo (Scale)

5,000

Peak connections (Scale)

99.999%

Uptime SLA (Enterprise)

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/jyswee/oddsockets-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, nil); 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/jyswee/oddsockets-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, nil); 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{
        Limit: 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"
    "encoding/json"
    "log"

    "github.com/gofiber/fiber/v2"
    "github.com/gofiber/fiber/v2/middleware/cors"
    "github.com/gofiber/fiber/v2/middleware/logger"
    "github.com/jyswee/oddsockets-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, nil); 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, then stream each message out as an SSE frame.
        // msg.Data is interface{} — encode it rather than casting.
        messages := make(chan *oddsockets.Message, 32)
        if err := ch.Subscribe(context.Background(), messages, nil); err != nil {
            return err
        }

        for msg := range messages {
            payload, err := json.Marshal(msg.Data)
            if err != nil {
                return err
            }
            if _, err := c.WriteString("data: " + string(payload) + "\n\n"); err != nil {
                return err
            }
        }

        return nil
    })
    
    log.Fatal(app.Listen(":8080"))
}

gRPC Integration

The request, response and stream types below come from your own protoc output, so generate them first — this contract is the minimum the example needs:

protobuf

// chat.proto
syntax = "proto3";
package chat;
option go_package = "your-app/gen/chatpb";

service ChatService {
  rpc SendMessage(SendMessageRequest) returns (SendMessageResponse);
  rpc Subscribe(SubscribeRequest) returns (stream MessageEvent);
}

message SendMessageRequest {
  string channel = 1;
  string message = 2;
  string user_id = 3;
}

message SendMessageResponse {
  bool success = 1;
}

message SubscribeRequest {
  string channel = 1;
}

message MessageEvent {
  string channel = 1;
  string message = 2;
  string user_id = 3;
  int64 timestamp = 4;
}

bash

protoc --go_out=. --go-grpc_out=. chat.proto

Then wire OddSockets into the generated service. Because this file imports the chatpb package you just generated, it is a fragment of your project rather than a standalone program:

go

package main

import (
    "context"
    "fmt"
    "log"
    "net"

    "google.golang.org/grpc"

    "github.com/jyswee/oddsockets-go-sdk/oddsockets"

    chatpb "your-app/gen/chatpb" // protoc output from chat.proto
)

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

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 *chatpb.SendMessageRequest) (*chatpb.SendMessageResponse, error) {
    channel := s.client.Channel(req.Channel)

    _, err := channel.Publish(ctx, map[string]interface{}{
        "text":   req.Message,
        "userId": req.UserId,
    }, nil)
    if err != nil {
        return nil, err
    }

    return &chatpb.SendMessageResponse{Success: true}, nil
}

func (s *ChatServer) Subscribe(req *chatpb.SubscribeRequest, stream chatpb.ChatService_SubscribeServer) error {
    channel := s.client.Channel(req.Channel)

    // Messages arrive on a channel you own; relay them onto the gRPC stream
    // until the client goes away.
    messages := make(chan *oddsockets.Message, 64)
    if err := channel.Subscribe(stream.Context(), messages, nil); err != nil {
        return err
    }

    for {
        select {
        case <-stream.Context().Done():
            return channel.Unsubscribe()
        case msg := <-messages:
            // msg.Data is interface{} — render it rather than casting.
            if err := stream.Send(&chatpb.MessageEvent{
                Channel:   req.Channel,
                Message:   fmt.Sprintf("%v", msg.Data),
                UserId:    msg.UserID,
                Timestamp: msg.Timestamp.Unix(),
            }); err != nil {
                log.Printf("Failed to send message: %v", err)
                return 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()
    chatpb.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/jyswee/oddsockets-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 — the SDK delivers straight into the job
    // queue and drops (with a warning) if the buffer is full, so no relay
    // goroutine is needed.
    channel := mp.client.Channel("work-queue")
    if err := channel.Subscribe(ctx, mp.jobQueue, nil); 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": msg.Data, // interface{} — pass it through as-is
        "processedBy":     workerID,
        "processedAt":     time.Now().Unix(),
    }

    if _, err := resultChannel.Publish(ctx, result, nil); 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 {}
}