server phoenix (AsyncAPI)

Generate a Phoenix application of the WebSocket and Kafka channels of an AsyncAPI document.

Usage

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

Description

apigenie-generator server phoenix generates from OpenAPI documents (see apigenie-generator server phoenix) 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 Mix project (Elixir 1.18 or later): the WebSocket channels on Phoenix 1.7 with Bandit, and the Kafka channels on brod and Broadway. 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 counterpart has the Kafka channels only, since Phoenix serves WebSocket channels as the application. The project depends on snappyer, a NIF that needs a C++ compiler, so that brod reads Snappy-compressed records (the default of Go’s franz-go producers); LZ4 and zstd records need lz4b or ezstd added by hand.

The generator writes the channels of servers with the ws/wss and kafka/kafka-secure protocols, and channels with a WebSocket or Kafka binding in documents without servers. Channels of other protocols (MQTT, AMQP, HTTP) are skipped and reported as info.

defmodule MyApp.PickupDesk do
  @behaviour CarpetExpressEventsWeb.Channels.PickupDesk.Handler

  alias CarpetExpressEvents.Models

  @impl true
  def request_pickup(%Models.PickupRequest{} = request, _session) do
    # the reply gets the request's requestId when it has none
    {:reply, %Models.PickupConfirmed{request_id: nil, pickup: book(request)}}
  end
end

defmodule MyApp.LabelFarm do
  @behaviour CarpetExpressEvents.Kafka.RequestLabelRender

  @impl true
  def render_label(message, _metadata) do
    # sent to carpetexpress.label-render.results with the request's correlation id
    {:reply, %CarpetExpressEvents.Messages.LabelRendered{key: message.key, payload: render(message.payload)}}
  end
end

# config/config.exs
config :carpet_express_events, CarpetExpressEventsWeb.Channels.PickupDesk.Handler,
  impl: MyApp.PickupDesk

The project, for the application carpet_express_events (named by app, --package-name or the document’s title):

File	Contents
`lib/<app>/models/*.ex`	the structs of the schemas the payloads use, decoded and encoded by `decode.ex` and `encode.ex` (as the OpenAPI models of `server phoenix`)
`lib/<app>/messages/*.ex`	per Kafka message, a struct (`payload`, `headers`, `key`) and the description of its record
`lib/<app>/kafka.ex`	the supervisor of the Kafka processes and a function per topic
`lib/<app>/kafka/<operation>.ex`	per sent operation, its send (and request) functions; per received operation, the behaviour of its callbacks
`lib/<app>/kafka/<operation>_consumer.ex`	per received operation, its Broadway pipeline
`lib/<app>_web/user_socket.ex`, `channels/*.ex`	the socket of the Phoenix clients, a Phoenix channel per WebSocket channel and the behaviour of its callbacks (`channels/<channel>/handler.ex`)
`lib/<app>_web/endpoint.ex`	the endpoint: the socket, then the plain WebSockets of the channels
`lib/<app>/runtime/`, `lib/<app>_web/runtime/`	the runtime of the Kafka processes and of the channels
`config/*.exs`, `mix.exs`, `.formatter.exs`, `.credo.exs`	the project; `config/runtime.exs` reads `PORT` and `KAFKA_BROKERS`

WebSocket channels, two transports. Each channel is a Phoenix channel: Phoenix clients join its topic on /socket/websocket, the address with its parameters as segments (/tracking/{shipmentId} is tracking:<shipmentId>), with the ws binding’s query as the join payload. The same channel is served as a plain WebSocket at its address, where the frames are the messages as the AsyncAPI document describes them (the prefix of <Root>Web.Runtime.PlainSocketPlug mounts the addresses under a path). Both transports decode the address parameters and the query, call the optional join/2 callback, decode received messages into their structs before the callbacks get them, and send what the channel module’s push_<message>/2 (one connection) and broadcast_<message> (every connection to the topic) functions send. A request’s callback returns {:reply, message}: the reply of the Phoenix protocol, or a frame on a plain WebSocket, with the request’s correlation id copied when it has none.
Received Kafka operations. A Broadway pipeline per operation reads its topic in its consumer group and hands each record to a callback of the operation’s behaviour (CarpetExpressEvents.Kafka.PublishShipmentEvents), named after the message. Several messages of a channel are told apart by a header or a payload property with a distinct constant value (ce_type), otherwise by the first message whose payload and headers decode. Records matching no message go to the optional on_unknown/1, records that don’t decode to the optional on_invalid/2 (and fail). Batching: batch_size and batch_timeout from x-apigenie-broadway of the operation’s (or channel’s) Kafka binding, else the broadway option, else Broadway’s 100 records and 1000 ms; the pipeline module’s configuration overrides them, and the consumer group and offset reset policy too.
Sent Kafka operations. send_<message>/1 checks the headers of a message, encodes it and returns once the brokers have the record, keyed by key.
Request/reply on Kafka, with a reply channel: request_<message>/2 sends the request with its correlation id (generated when it has none) and waits for the reply (:timeout, 30 seconds by default), read by a consumer of the reply topic without consumer group; a replier’s callback returns the reply, sent to the reply topic with the request’s correlation id.

Group ids come from the groupId schemas of the Kafka operation bindings (their const, single enum value, default or first example), else <app in kebab case>-<operation>. The partitions, replicas and topic configuration of channel bindings are documented on the topic functions; the generated code creates no topics.

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, Kafka topics with parameters, and WebSocket addresses whose segments mix text and parameters. Security is not generated: UserSocket.connect/3 accepts every connection and the Kafka clients connect without SASL or TLS. The client ids of the Kafka bindings are documented but not applied, the headers schema of the ws binding is not read, and the models decode by type: not, if/then/else, dependencies and number and array constraints aren’t checked.

Options

-c, –config

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

-o, –output

Directory where the project is generated.

–package-name

Name of the Mix application, converted to snake_case (carpet-express-events gives carpet_express_events, root module CarpetExpressEvents); defaults to the title of the API (events when it gives no name). Its kebab case is the base of default group ids.

–package-version

Version of the Mix project; 0.1.0 by default.

-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 phoenix; its other keys configure the generation from OpenAPI documents.

version: '1.0'
commands:
  generate:
    server_phoenix:
      asyncapi:
        app: carpet_express_events
        namespace: CarpetExpressEvents
        perspective: application
        broadway: { batch_size: 100, batch_timeout: 1000, concurrency: 10 }

Option	Type	Default	Description
`app`	string	from `--package-name` or the title	Name of the Mix application (snake_case)
`namespace`	string	the application in PascalCase	Root module of the generated modules; the web modules are in `<Root>Web`
`perspective`	`application` \| `counterpart`	`application`	The described application, or its counterpart, receiving what it sends
`field_prefix`	string	`field_`	Prefix of struct fields whose property names give no valid name
`module_overrides`	map		Module names of payload schemas, e.g. `ShipmentId: ShipmentRef`
`dev_deps`	boolean	`true`	credo and dialyxir as dependencies of the dev and test environments
`broadway`	map		`batch_size`, `batch_timeout` (ms) and `concurrency` of the pipelines