server go (AsyncAPI)

Generate a Go Kafka application of the Kafka channels of an AsyncAPI document.

Usage

apigenie-generator server go [...OPTIONS] INPUT_FILE

Description

apigenie-generator server go generates from OpenAPI documents (see apigenie-generator server go) and from AsyncAPI 3.x documents, telling them apart by their openapi or asyncapi field. Documents referencing other files are bundled first.

From an AsyncAPI document it generates a Go module (Go 1.22) using the Kafka client franz-go (github.com/twmb/franz-go/pkg/kgo, pinned at v1.18.1 in go.mod, with a complete go.sum). By default the application is the one the document describes: it sends the messages of send operations and receives the messages of receive operations. Set the perspective option to counterpart to swap them.

The generator writes the Kafka channels: the channels of servers with the kafka or kafka-secure protocol, and channels with a Kafka binding in documents without servers. Channels of other protocols are skipped and reported as info.

type labelFarm struct{}

func (labelFarm) OnRenderLabel(ctx context.Context, message events.RenderLabel) (events.RequestLabelRenderReply, error) {
	var reply events.LabelRendered
	reply.Key = message.Key
	reply.Payload = render(message.Payload)
	return reply, nil // sent to TopicLabelRenderResults with the request's correlation id
}

func (labelFarm) OnUnknown(ctx context.Context, record *kgo.Record) error { return nil }

func main() {
	var handlers events.Handlers
	handlers.RequestLabelRender = labelFarm{}
	// ...
	app, err := events.NewApp(events.Options{Brokers: []string{"localhost:9092"}}, handlers)
	if err != nil {
		log.Fatal(err)
	}
	defer app.Close()
	err = app.PublishShipmentEvents().SendShipmentDelivered(ctx, delivered)
	// ...
	_ = app.Run(ctx) // consumes until ctx is done
}

The module has one package at its root, named by package_name, --package-name or the document’s title (letters and digits, lower case):

File	Contents
`types.gen.go`	the types of the schemas the payloads use: structs with `json` tags, enums, and unions as a struct whose `Value` holds one variant of a sealed interface (`switch v := event.Value.(type)`); each with a `Validate` method
`messages.gen.go`	a struct per message (`Key`, `Headers`, `Payload`), its headers struct and header constants, and the code writing, reading and validating records
`kafka.gen.go`	the server hosts, topics, group and client ids, the handler interfaces, `Handlers`, `App` and the senders
`runtime_*.gen.go`	the Kafka runtime (consumers, requesters, headers), validation, `Nullable[T]`, `Date`
`go.mod`, `go.sum`	the module, requiring franz-go
`cmd/<package>/main.go`	a starting point, written once and never overwritten

Messages. Headers are the message’s headers merged with those of its traits as AsyncAPI does (JSON Merge Patch: a later required list replaces an earlier one). Header values are text on Kafka, so headers are string, int64, float64 or bool fields, pointers when optional. Headers with a constant value, e.g. the ce_specversion and ce_type of CloudEvents in binary mode, are constants (ShipmentStatusChangedCeType), which senders write and receivers compare. Messages are validated as a whole before they are sent and after they are received.
Received operations. NewApp(options, handlers) takes a handler per received operation (PublishShipmentEventsHandler) with an On<Message> method per message and OnUnknown. Several messages of a channel are told apart by a payload property or a header with a distinct constant value (e.g. type or ce_type); otherwise the first message whose payload decodes strictly and is valid is chosen. Records matching no message go to OnUnknown. Run consumes the operations’ topics, each in its consumer group, until its context is done; Close commits the offsets of the handled records and leaves the groups. Records that aren’t valid messages, and handler errors, go to Options.OnError (logged by default); the records are committed either way.
Sent operations. app.PublishShipmentEvents() returns the sender of the operation, with a Send<Message>(ctx, message) method per message that validates the message and returns once the brokers have it.
Request/reply with a reply channel: the requester’s Send<Message> returns the reply (one of the reply messages, RequestLabelRenderReply), matched by the request’s correlation id, which is generated when the request has none; Options.ReplyTimeout bounds the wait when the context has no deadline. A replier’s handler returns the reply, which is sent to the reply topic with the request’s correlation id copied (to a header, or a payload property, as the correlationId locations say).

Group and client ids come from the groupId and clientId schemas of the Kafka operation bindings, of the operation or its traits: their const, single enum value, default or first example. Without a value, <package name in kebab case>-<operation> is used, and the doc comment of the constant names the binding’s pattern. Options.GroupIDs replaces them by operation id. The partitions, replicas and topic configuration of channel bindings are documented on the topic constants; the generated code creates no topics.

main.go connects to the brokers of the -brokers flag, else of the KAFKA_BROKERS environment variable (comma-separated), else localhost:9092, wires stub handlers that return api.ErrNotImplemented, and stops cleanly on SIGINT or SIGTERM.

Not supported, and reported as warnings: Avro payloads and payloads of other non-JSON schema formats (the messages are skipped, and operations left without a message too), reply addresses given by runtime expressions, and Kafka topics with parameters. Security (SASL/SCRAM, mTLS) is not configured: pass kgo.SASL or kgo.DialTLSConfig in Options.ClientOptions. Validate doesn’t check not, if/then/else and dependencies, tuples are []any (only their item count is checked), and contentEncoding: base64 strings stay strings. With a correlation id in the payload at a place that isn’t a string field of the payload’s structs, the request must carry the id and the replier’s handler must copy it.

Extensions

Extension	Effect
`x-go-name`	name of the type of a schema, or of the field of a property
`x-go-type`	Go type of a schema, e.g. `decimal.Decimal`; with `x-go-type-import: {path: ...}` for its package
`x-go-type-skip-optional-pointer`	the optional property isn’t a pointer
`x-go-json-ignore`	the field gets the tag `json:"-"`
`x-omitempty`	`true` or `false` overrides the `omitempty` of the field
`x-enum-varnames`, `x-enumNames`	names of the enum constants
`x-enum-descriptions`	comments of the enum constants
`x-apigenie.codegen.name`	name of the type of a component schema, when it has no `x-go-name`
`x-apigenie.codegen.disabled`	`true` on a channel or an operation leaves it out (with the operations of a channel left out)

The extensions can live in an overlay for this target instead of in the document, e.g. go.overlay.yaml:

overlay: 1.0.0
info:
  title: Go types
  version: 1.0.0
actions:
  - target: $.components.schemas.Order
    update:
      x-go-name: PurchaseOrder
  - target: $.components.schemas.Order.properties.total
    update:
      x-go-type: decimal.Decimal
      x-go-type-import:
        path: github.com/shopspring/decimal

apigenie-generator server go asyncapi.yaml --overlay go.overlay.yaml -o out

Options

-c, –config

Path to the configuration file to use. See the “Configuration” section for more information.

–overlay OVERLAY

Path to an OpenAPI Overlay (1.0 to 1.2) applied to the input before generating, e.g. one setting the extensions of the “Extensions” section. Can be repeated: the overlays apply in order, after those listed under commands.generate.server_go.overlay in the configuration file (with paths relative to it). Warnings and errors about a value an overlay set point to the overlay. The overlays apply to the AsyncAPI document bundled with the files it references.

–strict

Makes an overlay action whose target selects nothing an error (exit status 2, nothing is generated) instead of a warning (OVERLAY:LINE:COL: action #N (TARGET) selects no nodes.).

-o, –output

Directory where the module is generated.

–package-name

Name of the Go package, as letters and digits in lower case (carpet-express-events gives carpetexpressevents); defaults to the title of the API (events when it gives no valid name). Its kebab case is the base of default group and client ids.

–package-version

Ignored: Go modules have no version in go.mod.

-p, –property

Applies to OpenAPI documents only; using it with an AsyncAPI document is an error.

Configuration

The options are read from the asyncapi section of the configuration of apigenie-generator server go; its other keys configure the generation from OpenAPI documents.

version: '1.0'
commands:
  generate:
    server_go:
      asyncapi:
        module_path: github.com/acme/events
        package_name: events
        initialisms: [SKU]
        main: true
        perspective: application

Option	Description
`module_path`	Module path of `go.mod` (default: `example.com/<package>`)
`package_name`	Name of the package, a lower-case Go identifier (default: from `--package-name` or the title)
`initialisms`	Words written in capitals in identifiers, besides the common ones (`ID`, `URL`, …)
`main`	`true` (default): write `cmd/<package>/main.go` when it doesn’t exist
`perspective`	`application` (default): the described application; `counterpart`: its counterpart, receiving what it sends