eurus

package module
v2.1.0 Latest Latest
Warning

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

 Go to latest Published: Jul 24, 2026 License: MIT Imports: 13 Imported by: 0

README

Eurus

A WebSocket API gateway for microservices. Routes messages to services based on path patterns.

Eurus lets you distribute WebSocket connections across multiple backend services. Instead of all connections going to one server, a gateway routes messages to different services based on the message path - similar to how HTTP API gateways work. Each connection gets pinned to a service instance that maintains its state. Services are Velaros routers, so you get bidirectional communication, pattern-based routing, and connection state management.

Go Reference Go Report Card GitHub release License Sponsor

Table of Contents

Features

  • 🌐 WebSocket Gateway - Route messages to backend services transparently
  • 🔍 Service Discovery - Services announce routes automatically on startup
  • 📌 Socket Pinning - Each connection stays with one service instance
  • ⚖️ Load Balancing - Least-connections distribution for new connections
  • 🔒 Public/Private Routes - PublicBind() for external, Bind() for internal
  • 🔌 Pluggable Transports - Local for development, NATS for production
  • 🚀 Horizontal Scaling - Run multiple instances seamlessly
  • 🎯 Velaros Native - Services are standard Velaros routers
  • ⚡ High Performance - Minimal routing overhead
  • 🧪 Production Ready - Race-tested with comprehensive test coverage

Installation

go get github.com/RobertWHurst/eurus/v2

For production with NATS:

go get github.com/nats-io/nats.go

Quick Start

Here's a minimal chat application showing how Eurus distributes WebSocket connections across microservices. The gateway receives WebSocket connections from clients and routes messages to backend services based on the message path.

Important: Gateway and services must use the same encoding middleware (JSON, MessagePack, or Protobuf).

Gateway

The gateway accepts WebSocket connections and routes messages to services:

package main

import (
    "context"
    "log"
    "net/http"
    "github.com/RobertWHurst/velaros"
    "github.com/RobertWHurst/velaros/middleware/json"
    "github.com/RobertWHurst/eurus/v2"
    "github.com/RobertWHurst/eurus/v2/transport/localtransport"
)

func main() {
    // Use local transport for development (NATS for production)
    transport := localtransport.New()

    // Connect the gateway. Connect blocks like http.ListenAndServe, so run
    // it in a goroutine; Close (or canceling the context) shuts it down.
    gateway := eurus.NewGateway("main-gateway", transport)
    go func() {
        if err := gateway.Connect(context.Background()); err != nil {
            log.Fatal(err)
        }
    }()
    defer gateway.Close()

    // Mount gateway on Velaros router - this is where the magic happens
    router := velaros.NewRouter()
    router.Use(json.Middleware())  // Must match what services use
    router.Use(gateway)             // Gateway becomes middleware

    // Serve WebSocket connections on /ws
    http.Handle("/ws", router)
    http.ListenAndServe(":8080", nil)
}
Service

Services handle messages using standard Velaros routers:

package main

import (
    "log"
    "github.com/RobertWHurst/velaros"
    "github.com/RobertWHurst/velaros/middleware/json"
    "github.com/RobertWHurst/eurus/v2"
    "github.com/RobertWHurst/eurus/v2/transport/localtransport"
)

func main() {
    router := velaros.NewRouter()
    router.Use(json.Middleware())

    // PublicBind exposes routes through the gateway
    router.PublicBind("/chat/join", func(ctx *velaros.Context) {
        var req struct {
            Username string `json:"username"`
            Room     string `json:"room"`
        }
        ctx.Unmarshal(&req)

        // Socket storage persists for the entire connection
        ctx.SetOnSocket("username", req.Username)
        ctx.SetOnSocket("room", req.Room)

        log.Printf("User %s joined room %s", req.Username, req.Room)
        ctx.Reply(map[string]string{
            "status": "joined",
            "room":   req.Room,
        })
    })

    router.PublicBind("/chat/message", func(ctx *velaros.Context) {
        var msg struct {
            Text string `json:"text"`
        }
        ctx.Unmarshal(&msg)

        // Retrieve stored user info
        username := ctx.MustGetFromSocket("username").(string)
        room := ctx.MustGetFromSocket("room").(string)

        log.Printf("[%s] %s: %s", room, username, msg.Text)

        // In production, broadcast to room members
        ctx.Reply(map[string]string{"status": "sent"})
    })

    // Regular Bind() creates internal-only routes
    router.Bind("/health", func(ctx *velaros.Context) {
        ctx.Reply(map[string]string{"status": "healthy"})
    })

    // Connect to same transport as gateway
    transport := localtransport.New()
    service := eurus.NewService("chat-service", transport, router)

    log.Println("Chat service starting...")
    log.Fatal(service.Listen(context.Background()))  // Announces routes and blocks
}
Client

JavaScript client connecting through the gateway:

const ws = new WebSocket('ws://localhost:8080/ws');

ws.onopen = () => {
    // Join a chat room
    ws.send(JSON.stringify({
        path: '/chat/join',
        id: 'msg-1',
        data: {
            username: 'Alice',
            room: 'general'
        }
    }));
};

ws.onmessage = (event) => {
    const msg = JSON.parse(event.data);

    if (msg.id === 'msg-1' && msg.data.status === 'joined') {
        // Now we can send messages
        ws.send(JSON.stringify({
            path: '/chat/message',
            id: 'msg-2',
            data: { text: 'Hello everyone!' }
        }));
    }
};

Note: Messages must include path for routing and id for request/response correlation. The gateway uses the path to route to the correct service, just like HTTP routing.

Core Concepts

Gateway

The gateway is middleware that routes WebSocket messages to your backend services. When a client sends a message, the gateway looks at the path and forwards it to the appropriate service - just like an HTTP reverse proxy, but for WebSocket.

What makes this powerful is that the gateway maintains a "virtual connection" to the service. The client has one WebSocket connection to the gateway, and the gateway manages the routing to services transparently. This abstraction is what enables horizontal scaling for WebSocket.

Service

Services are just Velaros routers that handle messages. The key innovation is the distinction between public and private routes:

  • PublicBind() - Routes accessible through the gateway (announced to gateways)
  • Bind() - Internal routes NOT accessible through gateway (kept private)

This separation lets you expose customer-facing APIs while keeping health checks, metrics, and admin endpoints private. It's the same pattern as having public and private methods in a class, but for network services.

Socket Pinning

Here's the crucial part: when a client connects and sends its first message, the gateway "pins" that connection to a specific service instance. Every message from that client will go to the same instance until the connection closes.

Why does this matter? Because it lets you store state on the connection - user sessions, authentication, conversation context - and know it will be there for every message. Without pinning, each message could hit a different instance and you'd lose all your state.

// First message: authenticate and store state
router.PublicBind("/auth/login", func(ctx *velaros.Context) {
    // Store on socket - persists for connection lifetime
    ctx.SetOnSocket("userID", authenticatedUserID)
    ctx.SetOnSocket("role", userRole)
})

// Subsequent messages: access stored state
router.PublicBind("/api/data", func(ctx *velaros.Context) {
    userID := ctx.MustGetFromSocket("userID").(string)
    role := ctx.MustGetFromSocket("role").(string)
    // User state available for all messages from this connection
})
Service Discovery

Service discovery happens automatically. When you start a service, it announces its public routes to all gateways. When you stop a service, the gateway discovers this on the next failed message delivery and removes it from routing.

This is intentionally simple - no complex consensus protocols or leader election. Services announce themselves, gateways track them, and failed deliveries trigger cleanup. It's eventual consistency, which works perfectly for most applications.

Encoding Middleware

Important: The gateway and all services must use the same encoding middleware. This is how the gateway extracts the message path for routing.

// Gateway and services must match
router.Use(json.Middleware())     // JSON encoding
// OR
router.Use(msgpack.Middleware())  // MessagePack encoding
// OR
router.Use(protobuf.Middleware()) // Protocol Buffers

Pick one encoding for your entire system. JSON is great for development, MessagePack for performance, and Protocol Buffers when you need schema validation.

Creating a Gateway

Setting up a gateway is simple - give it a name, a transport, and mount it on a Velaros router:

Basic Setup
import (
    "net/http"
    "github.com/RobertWHurst/velaros"
    "github.com/RobertWHurst/velaros/middleware/json"
    "github.com/nats-io/nats.go"
    "github.com/RobertWHurst/eurus/v2"
    "github.com/RobertWHurst/eurus/v2/transport/natstransport"
)

// Connect to NATS
nc, _ := nats.Connect("nats://localhost:4222")
transport := natstransport.New(nc)

// Create gateway
gateway := eurus.NewGateway("api-gateway", transport)
go func() {
    if err := gateway.Connect(context.Background()); err != nil {
        log.Fatal(err)
    }
}()
defer gateway.Close()

// Optionally wait until the gateway has announced itself before serving
<-gateway.Ready()

// Mount on router with encoding
router := velaros.NewRouter()
router.Use(json.Middleware())  // Must match service encoding
router.Use(gateway)

// Start accepting WebSocket connections
http.Handle("/ws", router)
http.ListenAndServe(":8080", nil)
Multiple Gateways

Run multiple gateways for high availability. Services announce to all gateways automatically:

// Gateway 1 - Port 8080
gateway1 := eurus.NewGateway("api-gateway", transport1)
router1 := velaros.NewRouter()
router1.Use(json.Middleware())
router1.Use(gateway1)
go http.ListenAndServe(":8080", router1)

// Gateway 2 - Port 8081
gateway2 := eurus.NewGateway("api-gateway", transport2)
router2 := velaros.NewRouter()
router2.Use(json.Middleware())
router2.Use(gateway2)
go http.ListenAndServe(":8081", router2)

Creating Services

Services are just Velaros routers. The key is using PublicBind() for routes you want accessible through the gateway:

Basic Service
import (
    "github.com/RobertWHurst/velaros"
    "github.com/RobertWHurst/velaros/middleware/json"
    "github.com/RobertWHurst/eurus/v2"
    "github.com/RobertWHurst/eurus/v2/transport/natstransport"
)

// Create router with handlers
router := velaros.NewRouter()
router.Use(json.Middleware())  // MUST match gateway encoding

// Public routes - accessible via gateway
router.PublicBind("/api/users/:id", func(ctx *velaros.Context) {
    userID := ctx.Params().Get("id")
    user := getUserByID(userID)
    ctx.Reply(user)
})

router.PublicBind("/api/posts/**", func(ctx *velaros.Context) {
    // Wildcard captures rest of path
    path := ctx.Path()  // e.g., "/api/posts/2024/11/my-post"
    ctx.Reply(getPost(path))
})

// Private routes - NOT accessible via gateway
router.Bind("/health", func(ctx *velaros.Context) {
    ctx.Reply(map[string]string{"status": "healthy"})
})

router.Bind("/metrics", func(ctx *velaros.Context) {
    ctx.Reply(getMetrics())
})

// Create and start service
nc, _ := nats.Connect("nats://localhost:4222")
transport := natstransport.New(nc)
service := eurus.NewService("api-service", transport, router)

// Optional: Target specific gateways
service.GatewayNames = []string{"public-gateway"}  // Default: all gateways

// Listen blocks until the context is canceled or Close is called
log.Fatal(service.Listen(context.Background()))
Scaling Services

Run multiple instances for load distribution:

# Terminal 1
go run service.go  # Instance with auto-generated ID

# Terminal 2
go run service.go  # Another instance

# Terminal 3
go run service.go  # Third instance

The gateway distributes new connections using least-connections load balancing. Each connection stays pinned to its assigned instance.

Transports

Transports handle communication between gateways and services. You have two options:

Local Transport

Perfect for development and testing - everything runs in a single process:

import "github.com/RobertWHurst/eurus/v2/transport/localtransport"

transport := localtransport.New()
// Share this instance between gateway and services

Use this for:

  • Local development
  • Unit tests
  • Simple applications that don't need distribution
  • Getting started quickly without NATS
NATS Transport

For production systems where gateways and services run on different machines:

import (
    "github.com/nats-io/nats.go"
    "github.com/RobertWHurst/eurus/v2/transport/natstransport"
)

nc, _ := nats.Connect("nats://nats-server:4222")
transport := natstransport.New(nc)

NATS gives you:

  • True distribution across machines
  • Automatic reconnection and failover
  • High availability with clustering
  • Battle-tested messaging infrastructure

Delivery details worth knowing at scale:

  • Messages that fit in a single NATS payload are delivered with one request/reply round trip; oversized payloads fall back to an acked chunk stream automatically.
  • Received messages are dispatched on per-socket workers: ordering is preserved per socket, and a slow socket never stalls the others.
  • Each socket has a bounded dispatch queue (DispatchQueueSize). A socket that overflows its queue is closed as a slow consumer rather than being allowed to back up the whole instance.
  • Liveness heartbeats are batched — one request per service→gateway pair per interval, regardless of connection count.

Quick NATS setup:

# Development
docker run -p 4222:4222 nats:latest

# Production cluster
nats-server --cluster nats://0.0.0.0:6222 \
            --routes nats://node1:6222,nats://node2:6222

Architecture

Message Flow

Here's what happens when a client sends a message:

  1. Client connects to /ws endpoint
  2. Velaros accepts the WebSocket connection
  3. Client sends: {path: "/chat/join", id: "1", data: {...}}
  4. Velaros decodes the message and extracts the path
  5. Gateway middleware sees "/chat/join" and finds a service that handles it
  6. Gateway creates a virtual connection to that service instance (first message only)
  7. Service receives the message and handles it with the matching route
  8. Service sends response back through gateway to client

The beauty is that the client just sees a single WebSocket connection, while behind the scenes you can have dozens of service instances handling different routes.

Load Balancing

The gateway uses least-connections load balancing - new connections go to the instance with the fewest active connections. Simple and effective:

Instance A: 5 connections
Instance B: 3 connections  ← New connection goes here
Instance C: 4 connections

As connections close, the distribution automatically rebalances. No configuration needed, it just works.

Failure Handling

Eurus keeps things simple with eventual consistency:

  • Service crashes? The gateway discovers this on the next message delivery and removes it
  • Gateway crashes? Clients reconnect to another gateway instance
  • Network issues? NATS handles reconnection automatically

For production, you'll want health checks to detect failures faster:

// Check service health periodically
ticker := time.NewTicker(30 * time.Second)
for range ticker.C {
    for _, service := range services {
        if !pingService(service) {
            gateway.RemoveService(service)
        }
    }
}

But honestly? The default behavior works fine for most applications. Services fail, gateways detect it, clients reconnect. Simple.

Advanced Usage

Selective Gateway Routing

Services can target specific gateways for multi-tenant or security-zoned deployments:

// Public-facing service
publicService := eurus.NewService("api", transport, publicRouter)
publicService.GatewayNames = []string{"public-gateway"}
go publicService.Listen(ctx)

// Internal admin service
adminService := eurus.NewService("admin", transport, adminRouter)
adminService.GatewayNames = []string{"internal-gateway"}
go adminService.Listen(ctx)

// Service accessible from both
sharedService := eurus.NewService("shared", transport, sharedRouter)
// Empty GatewayNames = announces to all gateways
go sharedService.Listen(ctx)
Connection Lifecycle

Use Velaros hooks for connection setup and teardown:

router.UseOpen(func(ctx *velaros.Context) {
    // Runs once when client connects
    ctx.SetOnSocket("connectedAt", time.Now())
    log.Printf("Client connected: %s", ctx.SocketID())
    ctx.Next()
})

router.UseClose(func(ctx *velaros.Context) {
    // Runs once when client disconnects
    duration := time.Since(ctx.MustGetFromSocket("connectedAt").(time.Time))
    log.Printf("Client %s disconnected after %v", ctx.SocketID(), duration)
})
Bidirectional Communication

Services can request data from clients:

router.PublicBind("/monitor/start", func(ctx *velaros.Context) {
    // Start monitoring loop
    ticker := time.NewTicker(5 * time.Second)
    defer ticker.Stop()

    for {
        select {
        case <-ticker.C:
            // Request metrics from client
            var metrics ClientMetrics
            err := ctx.RequestInto(MetricsRequest{
                Timestamp: time.Now().Unix(),
            }, &metrics)

            if err != nil {
                return  // Client disconnected
            }

            log.Printf("Client CPU: %.2f%%, Memory: %.2f%%",
                metrics.CPU, metrics.Memory)

        case <-ctx.Done():
            return  // Connection closed
        }
    }
})

Client handles server requests:

ws.onmessage = (event) => {
    const msg = JSON.parse(event.data);

    // Server requesting metrics
    if (msg.path === '/monitor/start') {
        ws.send(JSON.stringify({
            id: msg.id,  // Echo the ID for correlation
            data: {
                cpu: getCPUUsage(),
                memory: getMemoryUsage()
            }
        }));
    }
};

Testing

Testing with Eurus is straightforward - use the local transport to run everything in-process:

// start runs a gateway or service in the background and waits until it is
// ready. Both Gateway and Service expose Ready() for exactly this.
func startGateway(t *testing.T, gateway *eurus.Gateway) {
    t.Helper()
    ctx, cancel := context.WithCancel(context.Background())
    done := make(chan error, 1)
    go func() { done <- gateway.Connect(ctx) }()
    t.Cleanup(func() { cancel(); <-done })
    <-gateway.Ready()
}

func startService(t *testing.T, service *eurus.Service) {
    t.Helper()
    ctx, cancel := context.WithCancel(context.Background())
    done := make(chan error, 1)
    go func() { done <- service.Listen(ctx) }()
    t.Cleanup(func() { cancel(); <-done })
    <-service.Ready()
}

func TestServiceRouting(t *testing.T) {
    // Everything runs in-process for testing
    transport := localtransport.New()

    // Set up gateway
    gateway := eurus.NewGateway("test-gateway", transport)
    startGateway(t, gateway)

    // Set up service
    router := velaros.NewRouter()
    router.Use(json.Middleware())
    router.PublicBind("/test", func(ctx *velaros.Context) {
        ctx.Reply(map[string]string{"result": "success"})
    })

    service := eurus.NewService("test-service", transport, router)
    startService(t, service)

    // Verify the gateway can route to our service
    assert.True(t, gateway.CanServePath("/test"))
}

Run with race detector:

go test -race

Comparison with Zephyr

Eurus and Zephyr are sister projects - same architecture, different protocols:

Zephyr Eurus
Protocol HTTP WebSocket
Router Navaros Velaros
State Stateless Stateful
Load Balance Per request Per connection
Best For REST APIs Real-time

Use both together for complete coverage:

// HTTP API with Zephyr
httpGateway := zephyr.NewGateway("http-gateway", transport)
http.Handle("/api/", httpGateway)

// WebSocket with Eurus
wsGateway := eurus.NewGateway("ws-gateway", transport)
wsRouter := velaros.NewRouter()
wsRouter.Use(json.Middleware())
wsRouter.Use(wsGateway)
http.Handle("/ws", wsRouter)

http.ListenAndServe(":8080", nil)

Production Considerations

Observability

Add metrics for monitoring:

import "github.com/prometheus/client_golang/prometheus"

var (
    activeConnections = prometheus.NewGaugeVec(
        prometheus.GaugeOpts{
            Name: "eurus_active_connections",
            Help: "Active WebSocket connections",
        },
        []string{"gateway"},
    )

    messagesRouted = prometheus.NewCounterVec(
        prometheus.CounterOpts{
            Name: "eurus_messages_total",
            Help: "Total messages routed",
        },
        []string{"gateway", "service"},
    )
)
Health Monitoring

Implement health endpoints on services (using Bind() for internal access):

router.Bind("/health", func(ctx *velaros.Context) {
    health := CheckHealth()
    if !health.OK {
        ctx.Status = 503
    }
    ctx.Reply(health)
})

Help Welcome

If you want to support this project by throwing me some coffee money, it's greatly appreciated.

sponsor

If you're interested in providing feedback or would like to contribute, please feel free to do so. I recommend first opening an issue expressing your feedback or intent to contribute a change, from there we can consider your feedback or guide your contribution efforts. Any and all help is greatly appreciated since this is an open source effort after all.

Thank you!

License

MIT License - see LICENSE for details.

  • Velaros - WebSocket framework for Go (Eurus services use Velaros routers)
  • Zephyr - HTTP microservices gateway (sister project)
  • Navaros - HTTP framework for Go (Zephyr services use Navaros routers)
  • Conduit - Transport-agnostic messaging framework

Documentation

Index

Constants

View Source 
const ContextKeyRouteDescriptor = "eurus:route-descriptor"

ContextKeyRouteDescriptor is the key used to store the matched route descriptor on the velaros context.

View Source 
const ContextKeyServiceDescriptor = "eurus:service-descriptor"

ContextKeyServiceDescriptor is the key used to store the matched service descriptor on the velaros context.

View Source 
const ContextKeyServiceDescriptors = "eurus:service-descriptors"

ContextKeyServiceDescriptors is the key used to store all service descriptors on the velaros context.

Variables

View Source 
var (
	// ErrGatewayAlreadyConnected is returned by Gateway.Connect when the
	// gateway is already connected.
	ErrGatewayAlreadyConnected = errors.New("eurus: gateway already connected")

	// ErrServiceAlreadyListening is returned by Service.Listen when the
	// service is already listening.
	ErrServiceAlreadyListening = errors.New("eurus: service already listening")

	// ErrNoTransport is returned by Gateway.Connect and Service.Listen when no
	// transport was provided. Local services attached directly to a gateway
	// are driven by the gateway and must not be started on their own.
	ErrNoTransport = errors.New("eurus: no transport provided")

	// ErrSlowConsumer is returned by transport message delivery when the
	// receiving side rejected the message because the socket's dispatch queue
	// overflowed. It indicates a problem with the socket, not the receiving
	// instance.
	ErrSlowConsumer = errors.New("eurus: slow consumer")

	// ErrNoRouteMetadata is returned by RouteDescriptor.UnmarshalMetadata
	// when the route has no metadata attached.
	ErrNoRouteMetadata = errors.New("eurus: route has no metadata")
)
View Source 
var (
	ServicePruneInterval = 10 * time.Second
	ClosedSocketTTL      = 30 * time.Second
	HeartbeatInterval    = 5 * time.Second

	// ConnectionMailboxSize is the default number of inbound messages buffered
	// per connection before HandleMessage applies backpressure to the
	// transport's per-socket dispatch queue.
	ConnectionMailboxSize = 64
)
View Source 
var GatewayAnnounceInterval = time.Duration(8+rand.IntN(2)) * time.Second

GatewayAnnounceInterval is the default interval between gateway announcements. It is jittered per process so a fleet of gateways does not announce in lockstep.

View Source 
var GatewaySendTimeout = 30 * time.Second

GatewaySendTimeout is the default time allowed for delivering a single message to a client socket before the socket is considered failed.

Functions

This section is empty.

Types

type Connection 

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

func NewConnection 

func NewConnection(transport Transport, gatewayID, socketID string, info *velaros.ConnectionInfo, mailboxSize int, onClosed func()) *Connection

NewConnection creates a connection with a mailbox buffering up to mailboxSize inbound messages. When the mailbox is full HandleMessage blocks, applying backpressure to the transport's per-socket dispatch queue.

func (*Connection) Close 

func (c *Connection) Close(status velaros.Status, reason string) error

func (*Connection) GatewayID 

func (c *Connection) GatewayID() string

func (*Connection) HandleClose 

func (c *Connection) HandleClose(status velaros.Status, reason string)

func (*Connection) HandleMessage 

func (c *Connection) HandleMessage(msg *velaros.SocketMessage)

func (*Connection) Read 

func (c *Connection) Read(readCtx context.Context) (*velaros.SocketMessage, error)

func (*Connection) Write 

func (c *Connection) Write(ctx context.Context, msg *velaros.SocketMessage) error

type ConnectionMessage 

type ConnectionMessage struct {
	MessageType velaros.MessageType
	Data        []byte
}

type Gateway 

type Gateway struct {
	Name      string
	ID        string
	Transport Transport

	AnnounceInterval time.Duration
	SendTimeout      time.Duration
	// contains filtered or unexported fields
}

func NewGateway 

func NewGateway(name string, transport Transport) *Gateway

func (*Gateway) CanHandle 

func (g *Gateway) CanHandle(ctx *velaros.Context) bool

func (*Gateway) CanServePath 

func (g *Gateway) CanServePath(path string) bool

func (*Gateway) Close 

func (g *Gateway) Close() error

Close interrupts a running Connect call and waits for it to return. It is safe to call multiple times, and is a no-op when the gateway is not connected.

func (*Gateway) Connect 

func (g *Gateway) Connect(ctx context.Context) error

Connect joins the gateway to the transport and blocks until the context is canceled, Close is called, or a transport stream fails. A gateway may be connected again after a clean shutdown.

func (*Gateway) DescriptorMiddleware added in v2.1.0 

func (g *Gateway) DescriptorMiddleware() velaros.HandlerFunc

DescriptorMiddleware returns a velaros middleware that resolves the matching route descriptor for the incoming message and sets it on the context. Downstream middleware can retrieve it with RouteDescriptorFromContext — for example to enforce auth or rate-limit policy declared with velaros.WithMetadata before the message is dispatched.

func (*Gateway) Handle 

func (g *Gateway) Handle(ctx *velaros.Context)

func (*Gateway) HandleClose 

func (g *Gateway) HandleClose(ctx *velaros.Context)

func (*Gateway) Ready 

func (g *Gateway) Ready() <-chan struct{}

Ready returns a channel that is closed once the current Connect call has subscribed its transport streams and announced the gateway. It is intended for startup sequencing and readiness probes.

type GatewayDescriptor 

type GatewayDescriptor struct {
	Name               string               `msgpack:"name"`
	ServiceDescriptors []*ServiceDescriptor `msgpack:"serviceDescriptors"`
}

type GatewayServiceIndexer 

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

func NewGatewayServiceIndexer 

func NewGatewayServiceIndexer() *GatewayServiceIndexer

func (*GatewayServiceIndexer) Close 

func (r *GatewayServiceIndexer) Close()

func (*GatewayServiceIndexer) FreshServiceDescriptors 

func (r *GatewayServiceIndexer) FreshServiceDescriptors(threshold time.Duration) []*ServiceDescriptor

func (*GatewayServiceIndexer) IsClosed 

func (r *GatewayServiceIndexer) IsClosed() bool

func (*GatewayServiceIndexer) MapSocket 

func (r *GatewayServiceIndexer) MapSocket(serviceName, socketID string) (string, bool, error)

func (*GatewayServiceIndexer) PruneStaleServices 

func (r *GatewayServiceIndexer) PruneStaleServices(threshold time.Duration) ([]string, []string)

func (*GatewayServiceIndexer) Resolve added in v2.1.0 

func (r *GatewayServiceIndexer) Resolve(path string) (*ServiceDescriptor, *RouteDescriptor, bool)

Resolve finds the service and route that match the given path. It returns the matching service descriptor and route descriptor, or false when no service can handle the path.

func (*GatewayServiceIndexer) ResolveService 

func (r *GatewayServiceIndexer) ResolveService(path string) (string, bool)

ResolveService finds the name of the service that matches the given path.

func (*GatewayServiceIndexer) ServiceDescriptors 

func (r *GatewayServiceIndexer) ServiceDescriptors() []*ServiceDescriptor

ServiceDescriptors returns a snapshot of all current service descriptors.

func (*GatewayServiceIndexer) SetServiceDescriptor 

func (r *GatewayServiceIndexer) SetServiceDescriptor(descriptor *ServiceDescriptor) error

func (*GatewayServiceIndexer) SocketIDsByServiceInstance 

func (r *GatewayServiceIndexer) SocketIDsByServiceInstance() map[string][]string

func (*GatewayServiceIndexer) UnmapSocket 

func (r *GatewayServiceIndexer) UnmapSocket(socketID string)

func (*GatewayServiceIndexer) UnsetService 

func (r *GatewayServiceIndexer) UnsetService(id string) error

type RouteDescriptor 

type RouteDescriptor struct {
	Pattern *velaros.Pattern

	// Metadata carries arbitrary route metadata declared at the bind site
	// (for example with velaros.WithMetadata). After a msgpack round trip it
	// holds a msgpack.RawMessage; use UnmarshalMetadata to decode it into a
	// typed value regardless of how the descriptor was obtained.
	Metadata any
}

RouteDescriptor defines a route this service can handle. A route is a HTTP method, and a path matching pattern. It is used by the eurus gateway to determine which service to dispatch a request to.

func NewRouteDescriptor 

func NewRouteDescriptor(patternStr string) (*RouteDescriptor, error)

NewRouteDescriptor creates a new RouteDescriptor from a path pattern. The pattern determines which URL path this route will match.

func RouteDescriptorFromContext added in v2.1.0 

func RouteDescriptorFromContext(ctx *velaros.Context) *RouteDescriptor

RouteDescriptorFromContext retrieves the route descriptor that was set on the context by DescriptorMiddleware or Handle. Returns nil if no descriptor was set.

func (*RouteDescriptor) MarshalMsgpack 

func (r *RouteDescriptor) MarshalMsgpack() ([]byte, error)

MarshalMsgpack returns the msgpack representation of the route descriptor.

func (*RouteDescriptor) UnmarshalMetadata added in v2.1.0 

func (r *RouteDescriptor) UnmarshalMetadata(into any) error

UnmarshalMetadata decodes the route's metadata into the given value. It works both for descriptors received over a transport (where the metadata is a raw msgpack value) and for locally constructed descriptors (where it is the original value), so consumers behave identically across transports. Returns ErrNoRouteMetadata when the route has no metadata.

func (*RouteDescriptor) UnmarshalMsgpack 

func (r *RouteDescriptor) UnmarshalMsgpack(data []byte) error

UnmarshalMsgpack parses the msgpack representation of the route descriptor. Metadata is preserved as a raw msgpack value so consumers can decode it into their own types with UnmarshalMetadata.

type Service 

type Service struct {

	// GatewayNames is a list of gateway names that the service should announce
	// itself to. If the list is empty, the service will announce itself to all
	// gateways on the connection.
	GatewayNames []string

	// Name is the name of the service. This is used to identify the service
	// when announcing it to the gateway.
	Name string

	// ID is the unique identifier for the instance of the service. This is
	// automatically generated when the service is created.
	ID string

	// Transport is a struct that implements the Transport interface and
	// facilitates communication between services, gateways, and clients.
	Transport Transport

	// RouteDescriptors is a list of route descriptors that describe the routes
	// that this service can handle. If this is left empty, the service will
	// not be routable. This is automatically populated if the handler is a
	// Navaros router.
	RouteDescriptors []*RouteDescriptor

	// Router is called when a request is made to the service. This can be
	// either a Navaros router or a standard http.Router or http.HandlerFunc.
	Router *velaros.Router

	PruneInterval     time.Duration
	ClosedSocketTTL   time.Duration
	HeartbeatInterval time.Duration
	MailboxSize       int
	// contains filtered or unexported fields
}

Service is a struct that facilitates communication between a go microservice and a eurus gateway. It will manage the announcement of the service to the gateway as well as calls any HTTP Handler or Navaros handler.

If the service is given a Navaros router, it will automatically announce any public routes declared on the router. That said any handler compatible with go's http.HandlerFunc or http.Handler interface can be used.

Note that if you opt to use something other than Navaros, you will need to assign your route descriptors manually. This can be done by using the eurus.NewRouteDescriptor function to create your route descriptors, then assigning them to the RouteDescriptors field on the service.

func NewService 

func NewService(name string, transport Transport, handler any) *Service

NewService creates a new service with the given name, connection, and handler. The service will automatically announce itself to the gateway when it starts. If the handler is a Velaros router, the service will automatically announce any public routes declared on the router.

func (*Service) Close 

func (s *Service) Close() error

Close interrupts a running Listen call and waits for it to return. It is safe to call multiple times, and is a no-op when the service is not listening.

func (*Service) Listen 

func (s *Service) Listen(ctx context.Context) error

Listen starts the service and blocks until the context is canceled, Close is called, or a transport stream fails. A service may listen again after a clean shutdown.

func (*Service) Ready 

func (s *Service) Ready() <-chan struct{}

Ready returns a channel that is closed once the current Listen call has subscribed its transport streams and announced the service. It is intended for startup sequencing and readiness probes.

type ServiceDescriptor 

type ServiceDescriptor struct {
	Name             string             `msgpack:"name"`
	ID               string             `msgpack:"id"`
	GatewayNames     []string           `msgpack:"gatewayNames"`
	RouteDescriptors []*RouteDescriptor `msgpack:"httpRouteDescriptors"`

	LastSeenAt *time.Time `msgpack:"-"`
	// contains filtered or unexported fields
}

func ServiceDescriptorFromContext added in v2.1.0 

func ServiceDescriptorFromContext(ctx *velaros.Context) *ServiceDescriptor

ServiceDescriptorFromContext retrieves the service descriptor that was set on the context by DescriptorMiddleware. Returns nil if no descriptor was set.

func ServiceDescriptorsFromContext added in v2.1.0 

func ServiceDescriptorsFromContext(ctx *velaros.Context) []*ServiceDescriptor

ServiceDescriptorsFromContext retrieves the service descriptors that were set on the context by DescriptorMiddleware. Returns nil if no descriptors were set.

type Subscription 

type Subscription interface {
	Serve(ctx context.Context) error
}

Subscription is a live transport stream created by one of the Subscribe methods on Transport. By the time a Subscription is returned the stream is registered with the transport backend — messages published from this moment on will be observed.

Serve pumps the stream, invoking the handler passed to Subscribe. It must be called exactly once, blocks until the given context is canceled (returning nil) or the stream fails (returning the error), and releases the stream's resources before returning.

type SubscriptionFunc 

type SubscriptionFunc func(ctx context.Context) error

SubscriptionFunc adapts a function to the Subscription interface.

func (SubscriptionFunc) Serve 

func (f SubscriptionFunc) Serve(ctx context.Context) error

type Transport 

type Transport interface {
	// AnnounceGateway broadcasts a gateway descriptor to all services.
	AnnounceGateway(gatewayDescriptor *GatewayDescriptor) error
	// SubscribeGatewayAnnouncements subscribes to gateway announcements.
	SubscribeGatewayAnnouncements(ctx context.Context, handler func(gatewayDescriptor *GatewayDescriptor)) (Subscription, error)

	// AnnounceService broadcasts a service descriptor to all gateways.
	AnnounceService(serviceDescriptor *ServiceDescriptor) error
	// SubscribeServiceAnnouncements subscribes to service announcements.
	SubscribeServiceAnnouncements(ctx context.Context, handler func(serviceDescriptor *ServiceDescriptor)) (Subscription, error)

	// MessageService delivers a client message to a service instance. An error
	// indicates the message was not delivered and the caller may re-route.
	MessageService(serviceID, gatewayID, socketID string, connInfo *velaros.ConnectionInfo, msg *velaros.SocketMessage) error
	// SubscribeServiceMessages subscribes to messages addressed to a service
	// instance.
	SubscribeServiceMessages(ctx context.Context, serviceID string, handler func(gatewayID, socketID string, connInfo *velaros.ConnectionInfo, msg *velaros.SocketMessage)) (Subscription, error)

	// MessageGateway delivers a service message to the gateway that owns a
	// socket. An error indicates the message was not delivered.
	MessageGateway(gatewayID string, socketID string, msg *velaros.SocketMessage) error
	// SubscribeGatewayMessages subscribes to messages addressed to a gateway.
	SubscribeGatewayMessages(ctx context.Context, gatewayID string, handler func(socketID string, msg *velaros.SocketMessage)) (Subscription, error)

	// ClosedSocket broadcasts that a socket has been closed.
	ClosedSocket(socketID string, status velaros.Status, reason string) error
	// SubscribeSocketClosures subscribes to socket closure broadcasts.
	SubscribeSocketClosures(ctx context.Context, handler func(socketID string, status velaros.Status, reason string)) (Subscription, error)

	// HeartbeatSockets asks a gateway which of the given sockets are no longer
	// alive, and returns their IDs. An error indicates the gateway could not
	// be reached at all.
	HeartbeatSockets(gatewayID string, serviceID string, socketIDs []string) (dead []string, err error)
	// SubscribeSocketHeartbeats subscribes a gateway to heartbeat requests.
	// The handler returns the subset of socketIDs that are no longer alive.
	SubscribeSocketHeartbeats(ctx context.Context, gatewayID string, handler func(serviceID string, socketIDs []string) (dead []string)) (Subscription, error)
}

Transport moves announcements, socket messages, closures, and heartbeats between gateways and services.

Handler concurrency contract: announcement, closure, and heartbeat handlers may be invoked concurrently. Message handlers may be invoked concurrently for different sockets, but calls for the same socket are serialized in arrival order.

Source Files

View all Source files

Directories

Path Synopsis
transport
localtransport
natstransport

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.