Help us shape the future of Go! Take the 2026 Go Developer Survey

gateway

package module
v0.1.2 Latest Latest
Warning

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

 Go to latest Published: Apr 15, 2026 License: MIT Imports: 36 Imported by: 0

README

amqp-gateway

AMQP 0-9-1 HTTP Gateway — Publish messages to RabbitMQ via HTTP.

For clients that cannot connect to RabbitMQ directly, this gateway provides HTTP endpoints to publish messages and perform RPC calls over AMQP 0-9-1.

Usage

Server
$ amqp-gateway -c config.jsonnet
CLI Options
Usage: amqp-gateway [flags] <command>

Flags:
  -c, --config       Config file path (Jsonnet/JSON) [env: AMQP_GATEWAY_CONFIG]
      --log-level    Log level (debug, info, warn, error) [default: info, env: AMQP_GATEWAY_LOG_LEVEL]
      --version      Show version

Commands:
  run        Run the gateway server (default, requires --config)
  validate   Validate config (requires --config)
  render     Render config as JSON to stdout (requires --config)
  publish    Publish a message via HTTP
  rpc        Send an RPC request via HTTP
Configuration

Config files are evaluated by jsonnet-armed, supporting Jsonnet and JSON format. You can use Jsonnet features such as variables, functions, and environment variable expansion.

{
  amqp_url: "amqp://rabbitmq.internal:5672",
  listen_addr: ":8080",          // default: ":8080"
  shutdown_timeout: "30s",       // default: "30s"
  max_conns_per_user: 2,         // default: 2
  conn_ttl: "5m",                // default: "5m"
}

Environment variables can be referenced using must_env (fails if unset) or env (returns a default):

local must_env = std.native('must_env');
local env = std.native('env');
{
  amqp_url: must_env('AMQP_URL'),
  listen_addr: env('LISTEN_ADDR', ':8080'),
}
Key Description Default
amqp_url AMQP server URL (credentials are per-request via Basic auth) — (required)
listen_addr HTTP listen address :8080
shutdown_timeout Graceful shutdown timeout 30s
max_conns_per_user Max pooled AMQP connections per user/password/vhost 2
conn_ttl Max lifetime of a pooled connection before it is discarded 5m
aliases List of alias endpoint definitions (see below) []
Aliases

Aliases provide custom HTTP endpoints with pre-configured AMQP parameters. Clients can call these without knowing AMQP details or providing credentials.

{
  amqp_url: "amqp://localhost:5672",
  aliases: [
    {
      path: "/api/send-email",
      method: "publish",
      username: "app-user",
      password: "app-pass",
      exchange: "notifications",
      routing_key: "email.send",
      content_type: "application/json",
    },
    {
      path: "/api/user-lookup",
      method: "rpc",
      username: "app-user",
      password: "app-pass",
      routing_key: "user.lookup",
      timeout: "10s",
    },
  ],
}

# No auth needed — alias provides credentials
$ curl -X POST http://localhost:8080/api/send-email \
  -d '{"to": "user@example.com", "subject": "Hello"}'
# 202 Accepted

# RPC via alias
$ curl -X POST http://localhost:8080/api/user-lookup \
  -d '{"user_id": "123"}'
# 200 OK with response body

Alias fields:

Field Description Required
path HTTP path (e.g. /api/send) yes
method "publish" or "rpc" yes
username RabbitMQ username (if omitted, requires Basic auth in request) no
password RabbitMQ password (must be set together with username) no
exchange AMQP exchange no
routing_key AMQP routing key no
vhost AMQP vhost (default: /) no
delivery_mode 1=transient, 2=persistent (default: 2) no
content_type Default Content-Type no
timeout RPC timeout (default: 30s) no
headers Default AMQP headers (key-value map) no
response Custom HTTP response (see below) no

Response customization:

Field Description
response.status HTTP status code (overrides default: 202 for publish, 200 for rpc)
response.content_type Content-Type header (default: application/json)
response.body Response body. When content_type is text/*, must be a string and returned as-is. Otherwise JSON-encoded.

When response is set on an RPC alias, the RPC reply from RabbitMQ is discarded and the configured response is returned instead.

HTTP request headers (Amqp-*, Content-Type) can override alias defaults. If the client provides Authorization: Basic ..., it overrides the alias credentials.

Client Subcommands

publish and rpc subcommands act as HTTP clients to the gateway. They do not require --config.

# Publish a message
$ amqp-gateway publish http://localhost:8080 \
  -u guest -p guest \
  --routing-key my.queue \
  --content-type application/json \
  --body '{"hello":"world"}'

# Publish from file
$ amqp-gateway publish http://localhost:8080 \
  -u guest -p guest \
  --routing-key my.queue \
  --body-file message.json

# Publish from stdin
$ echo '{"hello":"world"}' | amqp-gateway publish http://localhost:8080 \
  -u guest -p guest \
  --routing-key my.queue \
  --body-file -

# RPC call
$ amqp-gateway rpc http://localhost:8080 \
  -u guest -p guest \
  --routing-key my.rpc.key \
  --timeout 5s \
  --body '{"request":"data"}'

# Custom AMQP headers
$ amqp-gateway publish http://localhost:8080 \
  -u guest -p guest \
  --routing-key my.queue \
  -H X-Request-Id=abc123 \
  --body test

--body and --body-file are mutually exclusive.

API

POST /v1/publish

Fire-and-forget publish with publisher confirm.

$ curl -X POST http://localhost:8080/v1/publish \
  -u guest:guest \
  -H "Content-Type: application/json" \
  -H "Amqp-Exchange: my-exchange" \
  -H "Amqp-Routing-Key: my.routing.key" \
  -d '{"message": "hello"}'
# 202 Accepted
POST /v1/rpc

RPC call — publishes a message and waits for a response on a temporary queue.

$ curl -X POST http://localhost:8080/v1/rpc \
  -u guest:guest \
  -H "Content-Type: application/json" \
  -H "Amqp-Exchange: my-exchange" \
  -H "Amqp-Routing-Key: my.rpc.key" \
  -H "Amqp-Timeout: 5000" \
  -d '{"request": "data"}'
# 200 OK with response body
GET /healthz

Service liveness check. Always returns 200 OK.

GET /readyz

RabbitMQ connectivity check. Returns 200 OK if connected, 503 otherwise.

HTTP Headers → AMQP Fields
HTTP Header AMQP Field Default
Amqp-Exchange exchange "" (default exchange)
Amqp-Routing-Key routing key ""
Amqp-Vhost vhost /
Amqp-Delivery-Mode delivery_mode 2 (persistent)
Amqp-Message-Id message_id auto-generated (UUID v7)
Amqp-Correlation-Id correlation_id auto-generated for RPC
Amqp-Expiration expiration —
Amqp-Timeout — 30000 (RPC only, ms)
Amqp-Header-* headers table —
Content-Type content_type —
Authentication

HTTP Basic authentication. Credentials are passed through to RabbitMQ for connection authentication.

Error Responses
Code Meaning
400 Validation error (invalid headers)
401 RabbitMQ authentication failure
403 Exchange permission denied
404 Exchange not found
504 RPC timeout
503 RabbitMQ unavailable

OpenTelemetry

amqp-gateway supports OpenTelemetry tracing and metrics. Enable by setting environment variables:

$ export OTEL_EXPORTER_OTLP_ENDPOINT=http://localhost:4318
$ export OTEL_EXPORTER_OTLP_PROTOCOL=http/protobuf  # or "grpc" (default: http/protobuf)
$ amqp-gateway -c config.jsonnet

When enabled, no code or config changes are needed — OTel is configured entirely via environment variables.

Traces

Trace context (W3C traceparent) flows through the full request lifecycle:

HTTP client → amqp-gateway → RabbitMQ → consumer (e.g. mqsubscriber)
  • HTTP request: Extracts traceparent from incoming HTTP headers
  • AMQP publish: Injects traceparent into AMQP message headers, so downstream consumers (e.g. mqsubscriber) can continue the trace
  • RPC response: Extracts traceparent from AMQP response and propagates it to the HTTP response

Spans created:

  • http.request — HTTP request lifecycle
  • amqp.publish — AMQP publish operation (SpanKind: Producer)
  • amqp.rpc — RPC operation (SpanKind: Client)
  • amqp.rpc.wait — Waiting for RPC response
Metrics
Metric Type Attributes
amqp_gateway.http.requests Counter method, path, status
amqp_gateway.http.duration Histogram (s) method, path
amqp_gateway.publish.total Counter result (success/error)
amqp_gateway.rpc.total Counter result (success/error/timeout)
amqp_gateway.log.messages Counter level
Log Correlation

When OTel is enabled, all structured logs (JSON) automatically include trace_id and span_id fields for correlation with traces.

Install

Homebrew
$ brew install fujiwara/tap/amqp-gateway
mise
$ mise use github:fujiwara/amqp-gateway
Go
$ go install github.com/fujiwara/amqp-gateway/cmd/amqp-gateway@latest
Binary

Download from Releases.

LICENSE

MIT

Author

fujiwara

Documentation

Index

Constants

View Source 
const (

	// AliasMethodPublish and AliasMethodRPC are the allowed values for AliasConfig.Method.
	AliasMethodPublish = "publish"
	AliasMethodRPC     = "rpc"
)

Variables

View Source 
var Version = "v0.1.2"

Functions

func NewServeMux 

func NewServeMux(client *AMQPClient, aliases []AliasConfig) http.Handler

NewServeMux creates the HTTP handler with all routes.

func RenderConfig 

func RenderConfig(ctx context.Context, path string) ([]byte, error)

RenderConfig evaluates the config file and returns pretty-printed JSON.

func RunCLI 

func RunCLI(ctx context.Context) error

RunCLI parses command-line arguments and executes the appropriate command.

func RunServer 

func RunServer(ctx context.Context, cfg *Config) error

RunServer starts the HTTP server.

Types

type AMQPClient 

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

AMQPClient handles AMQP connections and operations.

func NewAMQPClient 

func NewAMQPClient(cfg *Config, tracer trace.Tracer, metrics *Metrics) *AMQPClient

NewAMQPClient creates a new AMQPClient.

func (*AMQPClient) Close 

func (c *AMQPClient) Close()

Close closes all pooled connections.

func (*AMQPClient) Ping 

func (c *AMQPClient) Ping(ctx context.Context) error

Ping tests the RabbitMQ connection using the base URL (no credentials).

func (*AMQPClient) Publish 

func (c *AMQPClient) Publish(ctx context.Context, params *PublishParams, body []byte) error

Publish connects to RabbitMQ and publishes a message.

func (*AMQPClient) RPC 

func (c *AMQPClient) RPC(ctx context.Context, params *PublishParams, body []byte) (*amqp.Delivery, error)

RPC connects to RabbitMQ, publishes a message, and waits for a response.

type AliasConfig 

type AliasConfig struct {
	Path   string `json:"path"`
	Method string `json:"method"` // "publish" or "rpc"

	// AMQP parameters — same fields as PublishParams.
	// HTTP request headers can override these if provided.
	Username     string            `json:"username"`
	Password     string            `json:"password"`
	Exchange     string            `json:"exchange"`
	RoutingKey   string            `json:"routing_key"`
	VHost        string            `json:"vhost"`
	DeliveryMode *uint8            `json:"delivery_mode"`
	ContentType  string            `json:"content_type"`
	Timeout      time.Duration     `json:"timeout"`
	Headers      map[string]string `json:"headers"`

	// Response customization for alias endpoints.
	Response *AliasResponseConfig `json:"response,omitempty"`
}

AliasConfig defines a custom HTTP path that maps to a publish or rpc call with pre-configured AMQP parameters.

type AliasResponseConfig added in v0.0.3 

type AliasResponseConfig struct {
	Status      int    `json:"status,omitempty"`
	ContentType string `json:"content_type,omitempty"`
	Body        any    `json:"body,omitempty"`
}

AliasResponseConfig customizes the HTTP response for alias endpoints.

type CLI 

type CLI struct {
	Config   string           `kong:"short='c',env='AMQP_GATEWAY_CONFIG',help='Config file path (Jsonnet/JSON)'"`
	LogLevel string           `kong:"default='info',enum='debug,info,warn,error',env='AMQP_GATEWAY_LOG_LEVEL',help='Log level'"`
	Run      RunCmd           `cmd:"" default:"1" help:"Run the gateway server"`
	Validate ValidateCmd      `cmd:"" help:"Validate config"`
	Render   RenderCmd        `cmd:"" help:"Render config as JSON to stdout"`
	Publish  ClientPublishCmd `cmd:"" help:"Publish a message via HTTP"`
	RPC      ClientRPCCmd     `cmd:"" help:"Send an RPC request via HTTP"`
	Version  kong.VersionFlag `help:"Show version"`
}

CLI represents the command-line interface.

type ClientOptions 

type ClientOptions struct {
	URL           string            `kong:"required,arg,help='Gateway URL (e.g. http://localhost:8080)'"`
	User          string            `kong:"short='u',env='AMQP_GATEWAY_USER',help='Username for Basic auth'"`
	Password      string            `kong:"short='p',env='AMQP_GATEWAY_PASSWORD',help='Password for Basic auth'"`
	Exchange      string            `kong:"name='exchange',help='AMQP exchange'"`
	RoutingKey    string            `kong:"name='routing-key',help='AMQP routing key'"`
	VHost         string            `kong:"name='vhost',help='AMQP vhost'"`
	DeliveryMode  string            `kong:"name='delivery-mode',help='AMQP delivery mode (1=transient, 2=persistent)'"`
	MessageID     string            `kong:"name='message-id',help='AMQP message ID'"`
	CorrelationID string            `kong:"name='correlation-id',help='AMQP correlation ID'"`
	Expiration    string            `kong:"name='expiration',help='AMQP expiration'"`
	ContentType   string            `kong:"name='content-type',help='Content-Type header'"`
	Timeout       time.Duration     `kong:"name='timeout',help='RPC timeout (e.g. 5s, 500ms)'"`
	Headers       map[string]string `kong:"name='header',short='H',help='Custom AMQP headers (key=value)'"`
	Body          string            `kong:"name='body',xor='body',help='Message body string'"`
	BodyFile      string            `kong:"name='body-file',xor='body',help='Read message body from file (- for stdin)'"`
}

ClientOptions holds common options for publish and rpc client commands.

type ClientPublishCmd 

type ClientPublishCmd struct {
	ClientOptions `kong:"embed"`
}

ClientPublishCmd publishes a message via the gateway HTTP API.

func (*ClientPublishCmd) Run 

func (cmd *ClientPublishCmd) Run(ctx context.Context, cli *CLI) error

type ClientRPCCmd 

type ClientRPCCmd struct {
	ClientOptions `kong:"embed"`
}

ClientRPCCmd sends an RPC request via the gateway HTTP API.

func (*ClientRPCCmd) Run 

func (cmd *ClientRPCCmd) Run(ctx context.Context, cli *CLI) error

type Config 

type Config struct {
	AMQPURL         string        `json:"amqp_url"`
	ListenAddr      string        `json:"listen_addr"`
	ShutdownTimeout time.Duration `json:"shutdown_timeout"`
	MaxConnsPerUser int           `json:"max_conns_per_user"`
	ConnTTL         time.Duration `json:"conn_ttl"`
	Aliases         []AliasConfig `json:"aliases"`
}

Config represents the application configuration.

func LoadConfig 

func LoadConfig(ctx context.Context, path string) (*Config, error)

LoadConfig loads and parses a configuration file (Jsonnet or JSON).

type Metrics 

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

Metrics holds OpenTelemetry metric instruments.

type PublishParams 

type PublishParams struct {
	Username      string
	Password      string
	Exchange      string
	RoutingKey    string
	VHost         string
	DeliveryMode  uint8
	MessageID     string
	CorrelationID string
	Expiration    string
	ContentType   string
	Headers       amqp.Table
	Timeout       time.Duration
}

PublishParams holds the parsed AMQP parameters from HTTP headers.

func ParsePublishParams 

func ParsePublishParams(r *http.Request) (*PublishParams, error)

ParsePublishParams extracts AMQP parameters from HTTP request headers.

type RenderCmd 

type RenderCmd struct{}

RenderCmd renders the configuration as JSON.

func (*RenderCmd) Run 

func (cmd *RenderCmd) Run(ctx context.Context, cli *CLI) error

type RunCmd 

type RunCmd struct{}

RunCmd runs the gateway server.

func (*RunCmd) Run 

func (cmd *RunCmd) Run(ctx context.Context, cli *CLI) error

type TimeoutError 

type TimeoutError struct {
	Timeout fmt.Stringer
}

TimeoutError represents an RPC timeout.

func (*TimeoutError) Error 

func (e *TimeoutError) Error() string

type ValidateCmd 

type ValidateCmd struct{}

ValidateCmd validates the configuration file.

func (*ValidateCmd) Run 

func (cmd *ValidateCmd) Run(ctx context.Context, cli *CLI) error

type ValidationError 

type ValidationError struct {
	Field   string
	Message string
}

ValidationError represents a header validation error.

func (*ValidationError) Error 

func (e *ValidationError) Error() string

Source Files

View all Source files

Directories

Path Synopsis
cmd
amqp-gateway command

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.