🌊 stream - Real-time Communications for Go 1.25+

⚡ Quick Start

SSE (Server-Sent Events)

package main
import (
 "net/http"
 "time"
 "github.com/coregx/stream/sse"
)
func main() {
 http.HandleFunc("/events", func(w http.ResponseWriter, r *http.Request) {
 conn, _ := sse.Upgrade(w, r)
 defer conn.Close()
 ticker := time.NewTicker(1 * time.Second)
 defer ticker.Stop()
 for {
 select {
 case t := <-ticker.C:
 event := sse.NewEvent(t.Format(time.RFC3339)).WithType("time")
 conn.Send(event)
 case <-conn.Done():
 return
 }
 }
 })
 http.ListenAndServe(":8080", nil)
}

WebSocket

package main
import (
 "log"
 "net/http"
 "github.com/coregx/stream/websocket"
)
func main() {
 http.HandleFunc("/ws", func(w http.ResponseWriter, r *http.Request) {
 conn, _ := websocket.Upgrade(w, r, nil)
 defer conn.Close()
 for {
 msgType, data, err := conn.Read()
 if err != nil {
 break
 }
 conn.Write(msgType, data)
 }
 })
 log.Fatal(http.ListenAndServe(":8080", nil))
}

hub := websocket.NewHub()
go hub.Run()
defer hub.Close()
http.HandleFunc("/ws", func(w http.ResponseWriter, r *http.Request) {
 conn, _ := websocket.Upgrade(w, r, nil)
 hub.Register(conn)
 defer hub.Unregister(conn)
 for {
 _, data, _ := conn.Read()
 hub.Broadcast(data)
 }
})

🌟 Why stream?

Zero Dependencies, Maximum Control

// Pure stdlib - no external dependencies in production
import "github.com/coregx/stream/sse"
import "github.com/coregx/stream/websocket"

RFC-Compliant Protocols

Broadcasting Made Simple

// Hub pattern for efficient broadcasting
hub := websocket.NewHub()
hub.BroadcastText("Hello, everyone!")
hub.BroadcastJSON(Message{Type: "update", Data: "..."})

📦 Installation

go get github.com/coregx/stream

🚀 Features

SSE (Server-Sent Events)

WebSocket

Common Features

📚 Documentation

🎯 Use Cases

Real-time Dashboards (SSE)

// Push live metrics to dashboard
conn.Send(sse.NewEvent(metricsJSON).WithType("metrics"))

Chat Applications (WebSocket)

// Bidirectional messaging
hub.BroadcastText(fmt.Sprintf("%s: %s", username, message))

Live Notifications (SSE)

// Server pushes notifications
event := sse.NewEvent(notification).WithType("alert").WithRetry(3000)
conn.Send(event)

IoT & Sensors (WebSocket)

// Binary data streaming
conn.Write(websocket.BinaryMessage, sensorData)

📊 Benchmarks

SSE Performance

BenchmarkSSE_Send-8 50000 23.4 μs/op 0 allocs/op
BenchmarkSSE_Broadcast-8 30000 47.2 μs/op 1 allocs/op
BenchmarkSSE_E2E_Latency-8 20000 68.5 μs/op 2 allocs/op

WebSocket Performance

BenchmarkWS_Echo-8 100000 15.3 μs/op 0 allocs/op
BenchmarkWS_Broadcast-8 50000 32.1 μs/op 1 allocs/op
BenchmarkWS_Hub_1000clients-8 10000 156 μs/op 3 allocs/op

🔧 Advanced Usage

SSE with Custom Headers

conn, err := sse.Upgrade(w, r)
if err != nil {
 http.Error(w, err.Error(), http.StatusBadRequest)
 return
}
// Set custom retry interval
event := sse.NewEvent("data").WithRetry(5000) // 5 seconds
// Named event types
event = sse.NewEvent("update").WithType("user-joined")
// Event IDs for reconnection
event = sse.NewEvent("data").WithID("msg-123")

WebSocket with Configuration

opts := &websocket.UpgradeOptions{
 ReadBufferSize: 4096,
 WriteBufferSize: 4096,
 CheckOrigin: func(r *http.Request) bool {
 return r.Header.Get("Origin") == "https://example.com"
 },
}
conn, err := websocket.Upgrade(w, r, opts)

Graceful Shutdown

// SSE
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
select {
case <-ctx.Done():
 conn.Close()
case <-conn.Done():
 // Client disconnected
}
// WebSocket Hub
hub.Close() // Gracefully closes all connections

🤝 Sister Projects

📊 Status

📄 License

🙏 Acknowledgments

Special Thanks