Flutter Friendly: “🪙Привет, это Катя, Flutter Dev Friflex Сегодня я хочу рассказать вам…”

🪙Привет, это Катя, Flutter Dev Friflex

Сегодня я хочу рассказать вам о библиотеке dart_amqp — полнофункциональном клиенте для работы с протоколом AMQP (Advanced Message Queue Protocol). Эта библиотека позволяет приложениям взаимодействовать с брокерами сообщений, такими как RabbitMQ.

Что такое dart_amqp?
dart_amqp — это клиентская библиотека для работы с AMQP-серверами, которая предоставляет удобный API для создания распределенных систем обмена сообщениями. Она поддерживает все основные возможности протокола AMQP, включая очереди, обменники, подтверждения сообщений и транзакции.

Создание клиента
▪️Настройка подключения
Для тонкой настройки подключения используется класс ConnectionSettings, который позволяет переопределить параметры по умолчанию:

Client client = Client(
  settings: ConnectionSettings(
    host: "127.0.0.1",
    port: 5672,
    virtualHost: "/",
    authProvider: PlainAuthenticationProvider("guest", "guest"),
    maxConnectionAttempts: 1,
    reconnectWaitTime: Duration(milliseconds: 1500),
  ),
);

▪️Подключение к серверу
Клиент подключается к серверу лениво, но можно установить соединение явно:

await client.connect();

Аутентификация
Библиотека поставляется с двумя провайдерами аутентификации:
▫️PlainAuthenticationProvider — для простой аутентификации по логину и паролю
▫️AmqPlainAuthenticationProvider — альтернативный вариант Plain-аутентификации
▫️Можно создать собственный провайдер, реализовав интерфейс Authenticator

Работа с TLS
Для защищенных соединений можно передать SecurityContext:

Client client = Client(
  settings: ConnectionSettings(
    tlsContext: SecurityContext()
      ..setTrustedCertificates('path/to/cert.pem'),
    onBadCertificate: (certificate) => false,
  ),
);

Heartbeat
Heartbeat позволяет клиенту и серверу отслеживать активность соединения. Если обе стороны указывают ненулевой период (> 1 секунды), механизм активируется автоматически:

Client client = Client(
  settings: ConnectionSettings(
    tuningSettings: TuningSettings(
      heartbeatPeriod: const Duration(seconds: 60),
    ),
  ),
);
await client.connect();
print(client.tuningSettings.heartbeatPeriod);

Если сервер не отвечает в течение согласованного периода, выбрасывается исключение HeartbeatFailedException.

Работа с каналами
Каналы (Channels) — это виртуальные соединения внутри одного TCP-подключения:

Channel channel = await client.channel();

queue() — объявить именованную очередь
exchange() — объявить обменник для маршрутизации сообщений
select() — начать транзакцию
commit() — зафиксировать транзакцию
rollback() — откатить транзакцию

Работа с очередями

// Создание очереди
Queue queue = await channel.queue("my_queue");
// Публикация сообщения
queue.publish("Flutter Friendly");
// Потребление сообщений
Consumer consumer = await queue.consume();
consumer.listen((AmqpMessage message) {
  print("Получено: ${message.payloadAsString}");
  message.ack(); // Подтвердить обработку
});

Exchanges
Обменники позволяют маршрутизировать сообщения к нескольким получателям.

Exchange exchange = await channel.exchange(
  "my_exchange",
  ExchangeType.FANOUT,
);

// Публикация через обменник
exchange.publish("Broadcast message", "routing_key");

// Привязка потребителя
Consumer consumer = await exchange.bindPrivateQueueConsumer(
  ["routing_key"],
);

consumer.listen((AmqpMessage message) {
  print("Сообщение из exchange: ${message.payloadAsString}");
});

❓Если вам интересна эта тема, то в следующей статье я объясню работу с AmqpMessage, Publisher Confirms, обработку ошибок и реальные примеры работы с несколькими RabbitMQ инстансами одновременно.