RabbitMQ transport

The RabbitMQ transport makes Janus a client of a RabbitMQ broker. Janus consumes requests from a queue and publishes replies and events to an exchange. Use it when you build a service that sits between your clients and Janus, and your platform already runs on queues. The queue gives you delivery guarantees and lets the broker deal with a Janus restart.

Janus handles one application at a time: there is one request queue and one reply stream. Two applications that share the queues will see each other's replies.

Enable and configure

The transport reads janus.transport.rabbitmq.jcfg. It is off unless general.enabled (Janus API) or admin.admin_enabled (Admin API) is true.

general: {
	enabled = true
	json = "indented"            # indented, plain or compact (the default)
	host = "localhost"
	port = 5672
	#username = "guest"
	#password = "guest"
	#vhost = "/"
	to_janus = "to-janus"        # the queue Janus reads requests from
	from_janus = "from-janus"    # the routing key (and queue) of replies and events
	#queue_durable = true        # see the RabbitMQ 4 note below
	#declare_outgoing_queue = false
}
admin: {
	admin_enabled = true         # Admin API queues: to-janus-admin and from-janus-admin
}

The full table (exchanges, queue_name, heartbeat, retry interval, TLS) is in the transport files reference.

RabbitMQ 4

RabbitMQ 4 refuses the transient, non-exclusive queues that the C defaults declare (queue_durable = false). A default config therefore fails to start against RabbitMQ 4, in the C server and here alike. Set queue_durable = true, and declare_outgoing_queue = false if you create the reply queue yourself. This comes from PARITY_LOG.md (wave 102, RabbitMQ 4.4.0). The transcript below ran against RabbitMQ 3.13, where the defaults work.

Try it

The broker is rabbitmq:3-management in a Docker container. Requests are published with the management API (POST /api/exchanges/%2F/amq.default/publish, with the queue name as routing key and a correlation_id property), and replies are read from from-janus with POST /api/queues/%2F/from-janus/get. Any AMQP client does the same with basic_publish and basic_consume.

// Broker: RabbitMQ 3.13 (docker). The requests are published with the management API (default exchange, routing key = queue name). Janus replies to the from-janus queue and copies the correlation_id.

// ⇠ queues Janus declared at start
from-janus
from-janus-admin
to-janus
to-janus-admin

// → publish to queue to-janus
// correlation_id: corr-1
{
  "janus": "ping",
  "transaction": "r1"
}

// ← message on from-janus
// reply_to: Janus
// correlation_id: corr-1
// content_type: application/json
{
  "janus": "pong",
  "transaction": "r1"
}

// → publish to queue to-janus
// correlation_id: corr-2
{
  "janus": "create",
  "transaction": "r2"
}

// ← message on from-janus
// reply_to: Janus
// correlation_id: corr-2
// content_type: application/json
{
  "janus": "success",
  "transaction": "r2",
  "data": {
    "id": 1492126879001064
  }
}

// → publish to queue to-janus
// correlation_id: corr-3
{
  "janus": "attach",
  "session_id": 1492126879001064,
  "plugin": "janus.plugin.echotest",
  "transaction": "r3"
}

// ← message on from-janus
// reply_to: Janus
// correlation_id: corr-3
// content_type: application/json
{
  "janus": "success",
  "session_id": 1492126879001064,
  "transaction": "r3",
  "data": {
    "id": 649901008145518
  }
}

// A plugin message gets an ack and, later, the plugin's event. Match the event by `transaction` and `sender`, not by correlation_id: this build copies the request's correlation_id onto the event, while the C transport sends events without one.

// → publish to queue to-janus
// correlation_id: corr-4
{
  "janus": "message",
  "session_id": 1492126879001064,
  "handle_id": 649901008145518,
  "transaction": "r4",
  "body": {
    "audio": true
  }
}

// ← message on from-janus
// reply_to: Janus
// correlation_id: corr-4
// content_type: application/json
{
  "janus": "ack",
  "session_id": 1492126879001064,
  "transaction": "r4",
  "hint": "I'm taking my time!"
}

// ⇠ message on from-janus
// reply_to: Janus
// correlation_id: corr-4
// content_type: application/json
{
  "janus": "event",
  "session_id": 1492126879001064,
  "transaction": "r4",
  "sender": 649901008145518,
  "plugindata": {
    "plugin": "janus.plugin.echotest",
    "data": {
      "echotest": "event",
      "result": "ok"
    }
  }
}

// The Admin API uses its own queues. They exist because the config sets admin_enabled = true.

// → publish to queue to-janus-admin
// correlation_id: corr-5
{
  "janus": "list_sessions",
  "transaction": "r5",
  "admin_secret": "janusoverlord"
}

// ← message on from-janus-admin
// reply_to: Janus
// correlation_id: corr-5
// content_type: application/json
{
  "janus": "success",
  "transaction": "r5",
  "sessions": [
    1492126879001064
  ]
}

At start Janus declared the four queues of the config: to-janus, from-janus, and, because admin_enabled is set, to-janus-admin and from-janus-admin.

Reference

Queues and messages

What	Value
Request queue	to_janus (declared by Janus; with queue_name set, that queue is bound to janus_exchange by the to_janus routing key)
Reply and event routing key	from_janus, on janus_exchange (empty means the default exchange)
Reply queue	from_janus, declared by Janus unless declare_outgoing_queue = false
Admin API	to_janus_admin and from_janus_admin, with queue_name_admin, declare_outgoing_queue_admin and the other _admin keys
Body	one JSON document, in the format of json
Reply properties	content_type: application/json, reply_to: Janus, and the correlation_id of the request

Set correlation_id on the request: Janus copies it to the reply, so you can match replies to requests by it or by transaction.

Events

When you create a session on this transport, its events (plugin events, webrtcup, hangup, timeout) go to from_janus. In this build the event message carries the correlation id of the request that is current on the session (the transcript shows it on the event). In the C server events have no correlation id. Match events by transaction and sender.

Behaviour you may not expect

• verified The queues are declared as soon as Janus has connected, so a client can publish to them right away and the message waits if Janus is slow.
• verified The Admin API needs admin_secret in each request, as on every transport. The transcript used it.
• Janus keeps trying while the broker is down, with a delay that grows from 100 ms (multiplier 1.5) up to retry_interval. block_startup_until_connected is read but the other transports start without waiting for the broker.
• If the broker restarts, the link comes back by itself when heartbeat is above 0 (parity/status.json, checked live with a broker restart: with heartbeat 2 it reconnected, with 0 it did not, as in C).

Known gaps

• gap The reply carries the request's correlation id on pushed events, which C does not do (see above).
• gap block_startup_until_connected does not block startup.
• gap TLS (ssl_enabled, certificates) is ported and compared with C in the harness in 7 cases; I did not run it against a TLS broker.
• gap query_transport answers error 490.
• gap session created events from this transport carry no transport object. See event types.
• Runs on RabbitMQ 4 only with the config changes above. Not run on RabbitMQ 4 here.

Test it in Janus Bench

Janus Bench talks HTTP and WebSockets only. There is no check for this transport.