AsyncAPI Codegen

An AsyncAPI Golang Code generator that generates all Go code from the broker to the application/user. Just plug your application to your favorite message broker!

⚠️ We do our best to progressively satisfy the entire AsyncAPI specification,but some features may still be missing: please raise an issue on any bug ormissing feature.

❤️ Support is greatly appreciated and contributions are welcomed!

Inspired from popular deepmap/oapi-codegen

Contents

• Supported functionalities
• Usage
• Concepts
• Examples
• Supported Brokers
  • Kafka
  • NATS / NATS JetStream
  • Custom broker
• CLI options
• Advanced topics
  • Middlewares
  • Context
  • Logging
  • Versioning
  • Extensions
  • ErrorHandler
  • Validations
• Contributing and support

Supported functionalities

• AsyncAPI versions:
  • 2.6.0
  • 3.0.0
• Brokers:
  • Kafka
  • NATS / NATS JetStream
  • Custom
• Formats:
  • JSON
• Logging:
  • Elastic Common Schema (JSON)
  • Text (Humand readable)
  • Custom
• Others:
  • Versioning support

Usage

In order to use this library in your code, please execute the following lines:

# Install the tool
go install github.com/lerenn/asyncapi-codegen/cmd/asyncapi-codegen@latest
# Generate the code from the asyncapi file
asyncapi-codegen -i ./asyncapi.yaml -p <your-package> -o ./asyncapi.gen.go
# Install dependencies needed by the generated code
go get -u github.com/lerenn/asyncapi-codegen/pkg/extensions

You can also specify the generation part by adding a go generate instruction at the beginning of your file:

//go:generate go run github.com/lerenn/asyncapi-codegen/cmd/asyncapi-codegen@<version> -i ./asyncapi.yaml -p <your-package> -o ./asyncapi.gen.go

Docker image

You can also use the dockerized version of this tool:

docker run -v .:/code -w /code lerenn/asyncapi-codegen asyncapi-codegen -i ./asyncapi.yaml -p <your-package> -o ./asyncapi.gen.go

Concepts

Let's imagine a message broker centric architecture: you have the application that you are developing on the right and the potential user(s) on the left.

Being a two directional communication, both of them can communicate to each other through the broker. They can even communicate with themselves, in case of multiple users or application replication.

For more information about this, please refere to the official AsyncAPI concepts.

With Async API generated code

• Yellow parts: when using the codegen tool, you will generate the code that will act as an adapter (called controller) between the user, the broker, and the application.
• Red parts: you will need to fill these parts between user, broker and application. These will allow message production and reception with the generated code.
• Orange parts: these parts will be available in this repository if you use an already supported broker. However, you can also use the implement it yourself if the broker is not supported yet.

Examples

Here is a list of example, from basic to advanced ones.

Please note that the examples are separated in different subdirectories per broker.

It is strongly advised to read them in present order, following your AsyncAPI version.

• HelloWorld:
  • AsyncAPI v2
  • AsyncAPI v3
• Ping:
  • AsyncAPI v2
  • AsyncAPI v3

Supported Brokers

In order to connect your broker to the autogenerated code, you will need to create a controller that will be used to publish and subscribe to messages.

You can use one of the already supported brokers or implement your own.

Kafka

In order to use Kafka as a broker, you can use the following code:

broker, _:=kafka.NewController([]string{"<host>:<port>", /* additional hosts */}, /* options */)

Here are the options that you can use with the Kafka controller:

• WithGroupdID: specify the group ID that will be used by the controller. If not specified, default queue name (asyncapi) will be used.
• WithPartition: specify the partition that will be used by the controller. If not specified, default partition (0) will be used.
• WithMaxBytes: specify the maximum size of a message that will be received. If not specified, default value (10e6, meaning 10MB) will be used.
• WithLogger: specify the logger that will be used by the controller. If not specified, a silent logger is used that won't log anything.
• WithAutoCommit: specify if the broker should use auto-commit for incoming messages or manual commits. Note that commits are managed by the broker implementation regardless, with manual commits they are executed after the message is complete processed. Subscribers retain the option to manually handle errors via the ErrorHandler, to use mechanisms such as dead letter or retry topics. The default value is true
• WithSasl: specify sasl mechanism to connect to the broker. Per default no mechanism will be used.
• WithTLS: specify tls config to connect to the broker. Per default no tls config will be used.
• WithConnectionTest: specify if the controller should make a connection test on creation. The default value is true

Authentication and TLS

To use a TLS connection and or authentication for the connection to the kafka broker the following options can be used:

// Plain mechanismkafkaController, err:=kafka.NewController([]string{"<host>:<port>"},
kafka.WithGroupID(queueGroupID),
kafka.WithSasl(plain.Mechanism{Username: "<user>", Password: "<password>"}),
)
// Sha256 mechanismsha256Mechanism, err:=scram.Mechanism(scram.SHA256, "<user>", "<password>")
iferr!=nil {
// handle error
}
kafkaController, err:=kafka.NewController([]string{"<host>:<port>"},
kafka.WithGroupID(queueGroupID),
kafka.WithSasl(sha256Mechanism),
)
// Sha512 mechanismsha512Mechanism, err:=scram.Mechanism(scram.SHA512, "<user>", "<password>")
iferr!=nil {
// handle error
}
kafkaController, err:=kafka.NewController([]string{"<host>:<port>"},
kafka.WithGroupID(queueGroupID),
kafka.WithSasl(sha512Mechanism),
)
// TLS// configure tls.configmyTLSConfig:=&tls.Config{}
kafkaController, err:=kafka.NewController([]string{"<host>:<port>"},
kafka.WithGroupID(queueGroupID),
kafka.WithTLS(myTLSConfig),
)

NATS

In order to use NATS as a broker, you can use the following code:

// Create the NATS controllerbroker, _:=nats.NewController("nats://<host>:<port>")
deferbroker.Close()
// Add NATS controller to a new App controllerctrl, err:=NewAppController(broker, /* options */)
//...

Here are the options that you can use with the NATS controller:

• WithLogger: specify the logger that will be used by the controller. If not specified, a silent logger is used that won't log anything.
• WithQueueGroup: specify the queue group that will be used by the controller. If not specified, default queue name (asyncapi) will be used.
• WithConnectionOpts: specify connection Options for establishing connection with nats see Nats Options for more information. If not specified, no options will be used.

Authentication and TLS

To use a TLS connection and or authentication for the connection to the nats broker the following nats options can be used:

import (
"github.com/lerenn/asyncapi-codegen/pkg/extensions/brokers/nats"// import natsio go client option
natsio "github.com/nats-io/nats.go"
)
funcmain(){
myTLSConfig:=&tls.Config{}
natsController, err:=nats.NewController("nats://<host>:<port>",
nats.WithQueueGroup(queueGroupID),
nats.WithConnectionOpts(natsio.UserCredentials("<user.jwt>", "<user.nk>"), natsio.Secure(myTLSConfig))
)
}

NATS JetStream

In order to use NATS JetStream as a broker, you can use the following code:

// Create the NATS controllerbroker, _:=natsjetstream.NewController("nats://<host>:<port>", /* options */)
deferbroker.Close()
// Add NATS controller to a new App controllerctrl, err:=NewAppController(broker)
//...

It is important to either create/update a stream with WithStreamConfig or to use WithStream to specify the stream that will be used by the broker. Consumer for the user controller can be either created/updated with WithConsumerConfig or WithConsumer.

Limitations

• the messages will be ack'd from the consumer even though the subscription was not setup (this will be logged)

Custom broker

In order to connect your application and your user to your broker, we need to provide a controller to it. Here is the interface that you need to satisfy:

import(
"github.com/lerenn/asyncapi-codegen/pkg/extensions"
)
typeBrokerControllerinterface {
// Publish a message to the brokerPublish(ctx context.Context, channelstring, mw extensions.BrokerMessage) error// Subscribe to messages from the brokerSubscribe(ctx context.Context, channelstring) (msgschan extensions.BrokerMessage, stopchanany, errerror)
}

You can find that there is an extensions.BrokerMessage structure that is provided and that aims to abstract the event broker technology.

By writing your own by satisfying this interface, you will be able to connect your broker to the generated code.

CLI options

Generation parts (-g, --generate)

The default options for asyncapi-codegen will generate everything; user, application, and type definitions but you can generate subsets of those via the -generate flag. It defaults to user,application,types but you can specify any combination of those.

Here are the universal parts that you can generate:

• application: generate the application boilerplate. application requires the types in the same package to compile.
• user: generate the user boilerplate. It, too, requires the types to be present in its package.
• types: all type definitions for all types in the AsyncAPI spec. This will be everything under #components, as well as request parameter, request body, and response type objects.

Package name (-p, --package)

The package name is the name of the package that will be used in the generated code. It is important to have the same package name for the user, application, and types in order to compile the code.

Input files (-i, --input)

The input file is the path to the AsyncAPI specification file that will be used to generate the code. It can be either a JSON or a YAML file.

Also, you can specify dependencies by separating them with a comma:

asyncapi-codegen -i ./asyncapi.yaml,./dependency1.yaml,./dependency2.yaml -p <your-package> -o ./asyncapi.gen.go

A dependency can also be an OpenAPI document: only its components are kept, so its schemas can be referenced (e.g. $ref: './openapi.yaml#/components/schemas/Pet') even when the document contains fields AsyncAPI does not allow, such as an array-typed servers.

Output file (-o, --output)

The output file is the path to the file that will be generated by the tool. It will contain the generated code.

Disable formatting (-f, --disable-formatting)

By default, the generated code will be formatted using gofmt. If you want to disable this feature, you can use the -f flag.

JSON keys conversion (-c, --convert-keys)

By default, the generation will use the key specified in the asyncapi codegen. This is also called the none convertion. You can also convert your keys to snake, camel, or kebab.

Example

Given this schema:

Payload:
type: objectproperties:
This_is a-Property:
type: string

Here are the generated JSON sent, given by the different options:

• No conversion (none): { "This_is a-Property": "value" }
• Camel case (camel): { "thisIsAProperty": "value" }
• Pascal case (pascal): { "ThisIsAProperty": "value" }
• Kebab case (kebab): { "this-is-a-property": "value" }
• Snake case (snake): { "this_is_a_property": "value" }

Advanced topics

Middlewares

You can use middlewares that will be executing when receiving and publishing messages. You can add one or multiple middlewares using the WithMiddlewares function in the initialization of the App or User controller:

// Create a new app controller with middlewaresctrl, _:=NewAppController(/* Broker of your choice */, WithMiddlewares(myMiddleware1, myMiddleware2/*, ... */))

Here the function signature that should be satisfied:

func(ctx context.Context, msg*extensions.BrokerMessage, next extensions.NextMiddleware) error

Note: the returned context will be the one that will be passed to following middlewares, and finally to the generated code (and subscription callback).

Examples

Filtering messages

import(
"github.com/lerenn/asyncapi-codegen/pkg/extensions"// ...
)
funcmyMiddleware(ctx context.Context, _*extensions.BrokerMessage, _ middleware.Next) error {
// Execute this middleware only if this is a received messageextensions.IfContextValueEquals(ctx, extensions.ContextKeyIsDirection, "reception", func() {
// Do specific stuff if message is received
})
returnnil
}

You can even discriminate on more specification. Please see the Context section.

Modify messages before sending/receiving

import(
"github.com/lerenn/asyncapi-codegen/pkg/extensions"// ...
)
funcmyMiddleware(_ context.Context, msg*extensions.BrokerMessage, _ middleware.Next) error {
msg.Headers["additional"] ="some-info"returnnil
}

Stopping message processing

import(
"github.com/lerenn/asyncapi-codegen/pkg/extensions"// ...
)
funcmyMiddleware(_ context.Context, msg*extensions.BrokerMessage, _ middleware.Next) error {
ifmsg.Headers["author"] !="me" {
returnfmt.Errorf("this is not me, aborting...")
}
returnnil
}

Executing code after receiving/publishing the message

By default, middlewares will be executed right before the operation. If there is a need to execute code before and/or after the operation, you can call the next argument that represents the next middleware that should be executed or the operation corresponding code if this was the last middleware.

Here is an example:

import(
"github.com/lerenn/asyncapi-codegen/pkg/extensions"// ...
)
funcsurroundingMiddleware(ctx context.Context, next extensions.NextMiddleware) error {
// Pre-operationfmt.Println("This will be displayed BEFORE the reception/publication")
// Calling next middleware or reception/publication code// The given context will be the one propagated to other middlewares and operation source codeerr:=next(ctx)
// Post-operationfmt.Println("This will be displayed AFTER the reception/publication")
returnerr
}

Context

When receiving the context from generated code (either in subscription, middleware, logging, etc), you can get some information embedded in context.

To get these information, please use the functions from github.com/lerenn/asyncapi-codegen/pkg/extensions:

// Execute this middleware only if this is from "ping" channelextensions.IfContextValueEquals(ctx, extensions.ContextKeyIsChannel, "ping", func() {
// Do specific stuff if the channel is ping
})

You can find other keys in the package pkg/extensions.

Logging

You can have 2 types of logging:

• Controller logging: logs the internal operations of the controller (subscription, malformed messages, etc);
• Publication/Reception logging: logs every publication or reception of messages.

Controller logging

To log internal operation of the controller, the only thing you have to do is to initialize the controller with a logger, with the function WithLogger():

import(
"github.com/lerenn/asyncapi-codegen/pkg/extensions/brokers"// ...
)
funcmain() {
// Create a new app controller with an Elastic Common Schema JSON compatible loggerctrl, _:=NewAppController(/* Broker of your choice */, WithLogger(log.NewECS()))
// ...
}

You can find all loggers in the directory pkg/log.

By default, the provided loggers (NewText and NewECS) output every message. You can restrict them to a minimum severity level with the WithLevel option:

import(
log "github.com/lerenn/asyncapi-codegen/pkg/extensions/loggers"
)
// Only warnings and errors will be printed, info messages are discarded.logger:=log.NewECS(log.WithLevel(log.LevelWarning))

Available levels are LevelInfo (default, logs everything), LevelWarning (warnings and errors) and LevelError (errors only).

Publication/Reception logging

To log published and received messages, you'll have to pass a logger as a middleware in order to execute it on every published and received messages:

import(
"github.com/lerenn/asyncapi-codegen/pkg/extensions/brokers"// ...
)
funcmain() {
// Create a new app controller with a middleware for logging incoming/outgoing messagesloggingMiddleware:=middleware.Logging(log.NewECS())
ctrl, _:=NewAppController(/* Broker of your choice */, WithMiddlewares(loggingMiddleware))
// ...
}

Custom logging

It is possible to set your own logger to the generated code, all you have to do is to fill the following interface:

typeLoggerinterface {
// Info logs information based on a message and key-value elementsInfo(ctx log.Context, msgstring, info...log.AdditionalInfo)
// Error logs error based on a message and key-value elementsError(ctx log.Context, msgstring, info...log.AdditionalInfo)
}

Here is a basic implementation example:

typeSimpleLoggerstruct{}
func (loggerSimpleLogger) formatLog(ctx log.Context, info...log.AdditionalInfo) string {
varformattedLogInfostringfori:=0; i<len(keyvals)-1; i+=2 {
formattedLogInfo=fmt.Sprintf("%s, %s: %+v", formattedLogInfo, info.Key, info.Value)
}
returnfmt.Sprintf("%s, context: %+v", formattedLogInfo, ctx)
}
func (loggerSimpleLogger) Info(ctx log.Context, msgstring, info...log.AdditionalInfo) {
log.Printf("INFO: %s%s", msg, logger.formatLog(ctx, info...))
}
func (loggerSimpleLogger) Error(ctx log.Context, msgstring, info...log.AdditionalInfo) {
log.Printf("ERROR: %s%s", msg, logger.formatLog(ctx, info...))
}

You can then create a controller with a logger using similar lines:

// Create a new app controller with the custom loggerctrl, _:=NewAppController(
/* Broker of your choice */,
WithLogger(SimpleLogger{}), /* Use on as internal logger */WithMiddleware(middleware.Logging(SimpleLogger{})), /* Use to log incoming/outgoing messages */
)

The generated code passes the contextual information (channel, correlation ID, direction, provider, broker message, version) directly to your logger as explicit LogInfo arguments at every log call. Your logger therefore only has to log the msg and the info it is given — there is no need to unwrap any context key:

func (loggerSimpleLogger) Info(ctx context.Context, msgstring, info...extensions.LogInfo) {
// 'info' already contains the contextual values (channel, correlationID, …)// ... log msg with info
}

Migration note (from versions where loggers enriched from context): the contextual values are no longer read from the context.Context by the built-in Text/ECS loggers — they are now supplied as explicit LogInfo arguments by the generated code, using the generic keys returned by extensions.LogInfosFromContext (channel, correlationID, brokerMessage, direction, provider, version). This means:

• The built-in Text and ECS loggers now emit those generic keys instead of their previous custom names (e.g. ECS no longer maps to trace.id, event.action, event.original).
• Custom loggers receive everything as arguments and no longer need to call extensions.LogInfosFromContext themselves.

The values are still stored in the context.Context (it stays useful for other purposes), so a custom logger that wants to keep reading from context can still call extensions.LogInfosFromContext(ctx).

Versioning

If you are in need to do a migration or support multiple versions of your AsyncAPI specifications, you can use the versioning package:

import (
"github.com/lerenn/asyncapi-codegen/pkg/extensions/brokers/nats""github.com/lerenn/asyncapi-codegen/pkg/extensions/versioning"
v1 "path/to/asyncapi/spec/version/1"
v2 "path/to/asyncapi/spec/version/2"
)
funcmain() {
// Create a broker (here from NATS)broker, _:= nats.NewController("nats://nats:4222"))
deferbroker.Close()
// Add a version wrapper to the brokervw:=versioning.NewWrapper(broker)
// Create application for version 1appV1, _:=v1.NewAppController(vw, /* controller options */)
deferappV1.Close(context.Background())
// Create v2 appappV2, _:=v2.NewAppController(vw, /* controller options */)
deferappV2.Close(context.Background())
// ...
}

Then you can use each application independently:

err:=appV1.SubscribeHello(context.Background(), func(ctx context.Context, msg v1.HelloMessage) {
// Stuff for version 1
})
err:=appV2.SubscribeHello(context.Background(), func(ctx context.Context, msg v2.HelloMessage) {
// Stuff for version 2
})

That way, you can support multiple different versions with the same broker.

Version tagging

The versioning feature will add an application-version header to each message in order to have the correct version of the application on each of them.

Non-tagged messages

If messages can have no application-version, you can use the option WithDefaultVersion to add a default version to non-tagged messages.

vw:=versioning.NewWrapper(broker, versioning.WithDefaultVersion("1.1.4"))

Change header key for application version

Also, if you don't want to use this header as a recipient to the application version, you can specify your own header with the option WithVersionHeaderKey.

vw:=versioning.NewWrapper(broker, versioning.WithVersionHeaderKey("my-version-key"))

Knowing generated version

If you want to use the version of the AsyncAPI document used, you can access the constant AsyncAPIVersion that is generated with the types. It is generated as followed:

constAsyncAPIVersion="{{ .Info.Version }}"

Specification extensions

Operation Object extensions (AsyncAPI v3)

These extension properties apply to "Operation Objects" and override the Go name of the functions generated for that operation. When omitted, the default derived name is used.

• x-go-subscribe-func: overrides the SubscribeTo<Operation> method name.
• x-go-received-func: overrides the <Operation>Received subscriber callback name.
• x-go-reply-func: overrides the ReplyTo<Operation> method name.
• x-go-send-func: overrides the Send{As,To}<Operation> method name.
• x-go-request-func: overrides the Request{As,To}<Operation> method name.

For example,

operations:
ping:
action: receivechannel:
$ref: '#/channels/ping'x-go-subscribe-func: ListenForPingx-go-received-func: OnPingx-go-send-func: PublishPingx-go-request-func: AskPing

Schema Object extensions

These extension properties apply to "Schema Objects" in AsyncAPI spec.

• x-go-type: Overrides the default Go type with the specified Go type name.

  For example,

  schemas:
  Object:
  properties:
  flag:
  type: integerx-go-type: uint8

  will be generated as

  typeObjectstruct {
  Flaguint8`json:"flag"`
  }

• x-go-name: Overrides the Go identifier generated for a type or a struct field, while keeping the original property key in the JSON tag.

  For example,

  schemas:
  event:
  type: objectx-go-name: Eventproperties:
  event_id:
  type: integerx-go-name: EventID

  will be generated as

  typeEventstruct {
  EventID*int64`json:"event_id,omitempty"`
  }

• x-go-type-import: Specifies the import package for x-go-type. This has two properties name and path. path is the package import path, e.g. github.com/google/uuid. name is the package import name, which is optional. For example,

  schemas:
  Object:
  properties:
  flag:
  type: integerx-go-type: mypackage.Flagx-go-type-import:
  path: abc.xyz/repo/mypackage

  will be generated as

  import (
  "abc.xyz/repo/mypackage"
  )
  // ...typeObjectstruct {
  Flag mypackage.Flag`json:"flag"`
  }

  while

  schemas:
  Object:
  properties:
  flag:
  type: integerx-go-type: alias.Flagx-go-type-import:
  path: abc.xyz/repo/mypackagename: alias

  will be generated as

  import (
  alias "abc.xyz/repo/mypackage"
  )
  // ...typeObjectstruct {
  Flag alias.Flag`json:"flag"`
  }

• x-omitempty: Controls the addition of the omitempty tag in JSON tags of generated Go structures. When set to false, the omitempty tag won't be added even for optional fields.

  For example,

  schemas:
  User:
  type: objectproperties:
  id:
  type: stringname:
  type: stringx-omitempty: false

  will be generated as

  typeUserstruct {
  Id*string`json:"id,omitempty"`Name*string`json:"name"`
  }

Default values

When an optional property declares a default value, a SetDefaults() method is generated on the corresponding struct. Calling it fills the fields that are still unset (nil) with their default, leaving already-set fields untouched:

schemas:
Settings:
type: objectproperties:
retries:
type: integerdefault: 3

generates:

func (t*Settings) SetDefaults() {
ift.Retries==nil {
v:=int64(3)
t.Retries=&v
}
}

Defaults are supported for scalar properties (string, boolean, integer, number).

Multiple messages per operation (AsyncAPI v3)

When an operation (or its channel) declares more than one message, a dedicated send function is generated for each one, named after the operation and the message:

channels:
events:
address: app.eventsmessages:
userCreated: { $ref: '#/components/messages/userCreated' }userDeleted: { $ref: '#/components/messages/userDeleted' }operations:
sendEvents:
action: sendchannel: { $ref: '#/channels/events' }

generates SendAsSendEventsOperationForUserCreated and ...ForUserDeleted, each sending its own message type. Operations with a single message are unchanged.

On the receive side, a subscribe function is likewise generated per message, named after the operation and the message, with one callback per message type:

operations:
receiveEvents:
action: receivechannel: { $ref: '#/channels/events' }

generates SubscribeToReceiveEventsOperationForUserCreated(ctx, fn) and ...ForUserDeleted(ctx, fn) (plus the matching UnsubscribeFrom...For... functions and per-message methods on the subscriber interface). A single broker subscription is opened per channel and multiplexed to the registered handlers.

Each received message is discriminated with a try-each-until-valid strategy: it is unmarshalled and validated against every expected message schema (using go-playground/validator, so this dependency must be available in your module) and dispatched to the first one that matches. This is heuristic and can be ambiguous when several message schemas overlap — it works best when the messages have distinguishing required or enum/const fields. A message that matches none of the expected types is nacked and surfaced to the error handler as extensions.ErrNoMatchingMessage. Operations with a single message are unchanged.

Channel parameters from message location (AsyncAPI v3)

When a channel parameter declares a location runtime expression, the generated send functions auto-fill that parameter from the outgoing message when the caller leaves it empty. The parameter stays overridable: if you set it explicitly, your value is used as-is.

channels:
user:
address: app.user.{userId}parameters:
userId:
location: $message.payload#/userIdmessages:
userSignedUp:
$ref: '#/components/messages/userSignedUp'

With the spec above, SendAs... derives userId from msg.Payload.UserId, so the message is published on app.user.<userId> without passing the parameter manually. Supported locations are $message.payload#/... and $message.header#/..., and the targeted field must be a string.

ErrorHandler

You can use an error handler that will be executed when processing for messages failed. To add a custom ErrorHandler to your controller use the WithErrorHandler function in the initialization of the App or User controller:

// Create a new app controller with ErrorHandlerctrl, _:=NewAppController(/* Broker of your choice */, WithErrorHandler(myErrorHandler), ...)

Here the function signature that should be satisfied:

func(ctx context.Context, topicstring, msg*AcknowledgeableBrokerMessage, errerror)

Note: The default ErrorHandler is a Noop ErrorHandler doing nothing. By using a ErrorHandler you can add custom behavior for example to move messages to retry or dead letter topics/queues. Acks and Naks will be executed after the ErrorHandler, you can use the AcknowledgeableBrokerMessage in the handler to Ack/Nak the message manually.

Examples

Use the Logging ErrorHandler

// Create a new app controller with Logging ErrorHandlerctrl, _:=NewAppController(/* Broker of your choice */, WithErrorHandler(errorhandlers.Logging(mylogger)), ...)

Build a custom ErrorHandler and handle Ack/Nak of the message

func(ctx context.Context, topicstring, msg*extensions.AcknowledgeableBrokerMessage, errerror) {
// check error or move message to some other queue/topichandleTheErrorSomehow()
// Ack or Nak the messagemsg.Ack()
msg.Nak()
}

Validations

You can use go-playground/validator to validate the fields content against the contract.

The following tags are currently supported:

Asyncapi	Validator tag	Comment
required	required	For a full support, the flag --force-pointers is necessary
minLength	min
maxLength	max
minimum	gte
maximum	lte
exclusiveMinimum	gt
exclusiveMaximum	lt
uniqueItems	unique	Only for arrays
enum	oneof	Only string enum are supported

You can validate manually wherever you hold a generated payload, or plug in the provided opt-in middleware to reject invalid messages automatically:

import (
"github.com/go-playground/validator/v10""github.com/lerenn/asyncapi-codegen/pkg/extensions/middlewares"
)
ctrl, _:=NewUserController(broker, WithMiddlewares(
middlewares.Validation(validator.New(), func() any { return&UserMessagePayload{} }),
))

The middleware unmarshals the message payload and validates it against the generated tags; if validation fails it returns an error, so the message is neither delivered (on reception) nor sent (on publication).

Contributing and support

If you find any bug or lacking a feature, please raise an issue on the Github repository!

Also please do not hesitate to propose any improvment or bug fix on PR. Any contribution is warmly welcomed!

And if you find this project useful, please support it through the Support feature on Github.