GRPCQueue - Async gRPC over Message Queues

GRPCQueue transforms synchronous gRPC calls into asynchronous queue-based messages, enabling reliable fire-and-forget RPC patterns with any message queue backend.

Overview

GRPCQueue allows you to: - Make gRPC calls that are queued instead of executed immediately - Use any message queue (SQS, RabbitMQ, Kafka, Redis) as transport - Maintain standard gRPC interfaces and generated client code - Add reliability, buffering, and decoupling to microservices - Process messages concurrently with configurable workers

Inspiration: Based on ppg/grpc-queue

Features

• Transport Agnostic: Works with any message queue via pluggable Transport interface
• Standard gRPC: Use generated client code without modifications
• FIFO Support: Optional message grouping for ordered processing
• Interceptors: Full support for gRPC client and server interceptors (auth, logging, tracing)
• Error Handling: Automatic Ack/Nack with retry support
• Concurrency: Process multiple messages in parallel with errgroup
• Type Safety: Uses Protocol Buffers for message serialization

Usage

Producer (Client Side)

The Producer implements grpc.ClientConnInterface, making it a drop-in replacement for grpc.ClientConn.

import (
    "github.com/Aryon-Security/grpcqueue"
    pb "your/proto/package"
)

// Create transport that implements grpcqueue.Transport
transport := myCustomTransport

// Create producer with options
producer := grpcqueue.NewProducer(
    transport,
    grpcqueue.WithInterceptor(yourClientInterceptor),
    grpcqueue.WithMessageGroupIDExtractor(func(ctx context.Context, msg proto.Message) (string, error) {
        // Extract grouping key for FIFO queues
        return extractTenantID(msg), nil
    }),
)
defer producer.Close()

// Use with generated gRPC client (no code changes needed!)
client := pb.NewYourServiceClient(producer)
err := client.YourMethod(ctx, &pb.YourRequest{...})

Consumer (Server Side)

The Consumer implements grpc.ServiceRegistrar, allowing you to register standard gRPC service implementations.

import (
    "github.com/Aryon-Security/grpcqueue"
    pb "your/proto/package"
)

// Create transport
transport := myCustomTransport

// Create consumer with options
consumer := grpcqueue.NewConsumer(
    transport,
    grpcqueue.WithServerInterceptor(yourServerInterceptor),
)

// Register your gRPC service implementation (standard code!)
pb.RegisterYourServiceServer(consumer, &YourServiceImpl{})

// Start consuming messages
ctx := context.Background()
if err := consumer.Serve(ctx); err != nil {
    log.Fatal(err)
}

Transport Interface

GRPCQueue uses a simple, pluggable Transport interface:

type Transport interface {
    Sender
    Receiver
    io.Closer
}

type Sender interface {
    Send(ctx context.Context, msgs ...*Message) error
}

type Receiver interface {
    Receive(ctx context.Context) (<-chan *Message, error)
}

Custom Transports

Implement the Sender and Receiver interfaces for custom message queues:

type MyCustomTransport struct {
    // your queue client
}

func (t *MyCustomTransport) Send(ctx context.Context, msgs ...*grpcqueue.Message) error {
    // Send messages to your queue
    return nil
}

func (t *MyCustomTransport) Receive(ctx context.Context) (<-chan *grpcqueue.Message, error) {
    // Return channel that emits messages
    ch := make(chan *grpcqueue.Message)
    go t.pollMessages(ctx, ch)
    return ch, nil
}

func (t *MyCustomTransport) Close() error {
    // Cleanup
    return nil
}

Message Format

Messages are serialized as JSON using Protocol Buffers' standard JSON mapping:

{
  "@type": "type.googleapis.com/infra.QueueItem",
  "service": "pkg.YourService",
  "method": "YourMethod",
  "payload": {
    "@type": "type.googleapis.com/pkg.YourRequest",
    "field1": "value1",
    "field2": 42
  }
}

The payload uses google.protobuf.Any for type-safe deserialization.

FIFO Queues and Message Grouping

For FIFO queues (like AWS SQS FIFO), use WithMessageGroupIDExtractor:

producer := grpcqueue.NewProducer(
    transport,
    grpcqueue.WithMessageGroupIDExtractor(func(ctx context.Context, msg proto.Message) (string, error) {
        // Group by tenant for per-tenant ordering
        if req, ok := msg.(*pb.TenantRequest); ok {
            return req.TenantId, nil
        }
        return "", nil
    }),
)

The group ID is set as metadata with key x-message-group-id (AWS SQS compatible).

Interceptors

GRPCQueue fully supports gRPC interceptors for cross-cutting concerns:

// Client interceptor (auth, retries, logging)
clientInterceptor := func(ctx context.Context, method string, req, reply interface{},
    cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
    // Add authentication, logging, etc.
    log.Printf("Sending %s", method)
    return invoker(ctx, method, req, reply, cc, opts...)
}

producer := grpcqueue.NewProducer(
    transport,
    grpcqueue.WithInterceptor(clientInterceptor),
)

// Server interceptor (auth validation, logging, metrics)
serverInterceptor := func(ctx context.Context, req interface{},
    info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) {
    // Validate auth, log requests, etc.
    log.Printf("Processing %s", info.FullMethod)
    return handler(ctx, req)
}

consumer := grpcqueue.NewConsumer(
    transport,
    grpcqueue.WithServerInterceptor(serverInterceptor),
)

Metadata is automatically propagated through the queue.

Error Handling

• Producer: Returns errors immediately if queue publish fails
• Consumer: Automatically Nacks failed messages for retry
• Panics: Recovered and converted to Nacks
• Context Cancellation: Gracefully stops processing in-flight messages

func (s *YourService) YourMethod(ctx context.Context, req *pb.Request) (*pb.Response, error) {
    // If this returns an error, message is Nacked for retry
    // If this panics, message is Nacked
    // If context is canceled, processing stops gracefully
    return processRequest(req)
}

Testing

Create an in-memory transport for testing:

// Create in-memory transport
transport := NewInMemoryTransport()

// Use in tests
producer := grpcqueue.NewProducer(transport)
consumer := grpcqueue.NewConsumer(transport)

Limitations

• Unary RPCs Only: Streaming RPCs are not supported (returns error)
• Fire-and-Forget: Producer doesn't receive responses (replies are ignored)
• No Load Balancing: Use queue consumer groups for horizontal scaling
• Message Size: Limited by underlying queue.

Performance Considerations

• Concurrency: Consumer uses errgroup to process messages in parallel
• Batching: Some transports support batch sends (check transport docs)
• Latency: Adds queue latency (~10-100ms depending on transport)
• Throughput: Limited by queue throughput (SQS: 3000 msg/s, Kafka: much higher)

Dependencies

• google.golang.org/grpc - gRPC framework
• google.golang.org/protobuf - Protocol Buffers
• golang.org/x/sync/errgroup - Concurrent message processing

Contributing

See CONTRIBUTING.md for guidelines.

License

MIT License