Дружим RabbitMQ и Flutter/Dart

Future<Exchange> _initClient() async {   try {     /// Настройки подключения     final settings = ConnectionSettings(       // Если у вас RabbitMQ запущен на на локальном компьютере       // Укажите нужный IP-адрес или hostname       host: 'localhost',       // Порт       port: 5672,       // Данные для входа в RabbitMQ       authProvider: const PlainAuthenticator('guest', 'guest'),     );      /// Создаем клиента для подключения к RabbitMQ     Client client = Client(settings: settings);     // Подключаемся     await client.connect();     // Создаем канал, по которому будем отправлять сообщения     final channel = await client.channel();     // Декларируем exchange. Если он уже создан, то повторная декларация     // должна совпадать по параметрам.     // Если не совпадает — будет ошибка.     // Если exchange не существует, то он будет создан     final exchange = await channel.exchange(       'delayed_exchange',       // Используем ExchangeType типа x-delayed-message       ExchangeType.custom('x-delayed-message'),       durable: true,       arguments: {         'x-delayed-type': 'direct',       },     );      log('Успешно подключились к RabbitMQ');     return exchange;   } on Object catch (error, stackTrace) {     log('Ошибка подключения к RabbitMQ', error: error, stackTrace: stackTrace);     rethrow;   } } 

  void _sendMessage() {     // Публикуем сообщение. Ставим задержку 5000 мс (5 секунд) для примера     // Через 5 секунд сообщение будет доставлено в очередь из Exchange x-delayed-message     final int delayMs = 5000;     // Обязательно передаем задержку в хедере     final headers = {'x-delay': delayMs};     // Создаем сообщение которое будем отправлять, можно отправить объект     final message =         'Привет, я отложенное сообщение, которое было отправлено в ${DateTime.now()}';     // Настройки сообщения     final properties = MessageProperties()..headers = headers;      // Ключ очереди     final routingKey = 'messages_queue';     // Отправляем     exchange.publish(       message,       routingKey,       properties: properties,     );     log('Отправили сообщение с задержкой $delayMs мс');   }

import 'dart:developer';  import 'package:dart_amqp/dart_amqp.dart'; import 'package:flutter/material.dart';  void main() async {   /// Инициализируем клиента   final exchange = await _initClient();    runApp(MyApp(exchange: exchange)); }  Future<Exchange> _initClient() async {   try {     /// Настройки подключения     final settings = ConnectionSettings(       // Если у вас RabbitMQ запущен на не локальном компьютере       // укажите нужный IP-адрес или hostname       host: 'localhost',       // Порт       port: 5672,       // Данные для входа в RabbitMQ       authProvider: const PlainAuthenticator('guest', 'guest'),     );      /// Создаем клиента для подключения к RabbitMQ     Client client = Client(settings: settings);     // Подключаемся     await client.connect();     // Создаем канал, по которому будем отправлять сообщения     final channel = await client.channel();     // Декларируем exchange. Если он уже создан, то повторная декларация     // должна совпадать по параметрам.     // Если не совпадает — будет ошибка.     // Если exchange не существует, то он будет создан     final exchange = await channel.exchange(       'delayed_exchange',       // Используем ExchangeType типа x-delayed-message       ExchangeType.custom('x-delayed-message'),       durable: true,       arguments: {         'x-delayed-type': 'direct',       },     );      log('Успешно подключились к RabbitMQ');     return exchange;   } on Object catch (error, stackTrace) {     log(       'Ошибка подключения к RabbitMQ',       error: error,       stackTrace: stackTrace,     );     rethrow;   } }  class MyApp extends StatelessWidget {   // Exchange для публикаций   final Exchange exchange;    const MyApp({super.key, required this.exchange});    @override   Widget build(BuildContext context) {     return MaterialApp(       home: Scaffold(         appBar: AppBar(title: const Text('Flutter Producer Пример')),         body: Center(           child: ElevatedButton(             onPressed: _sendMessage,             child: Text('Отправить'),           ),         ),       ),     );   }    void _sendMessage() {     // Публикуем сообщение. Ставим задержку 5000 мс (5 секунд) для примера     // Через 5 секунд сообщение будет доставлено в очередь из Exchange x-delayed-message     final int delayMs = 5000;     // Обязательно передаем задержку в хедере     final headers = {'x-delay': delayMs};     // Создаем сообщение которое будем отправлять, можно отправить объект     final message =         'Привет, я отложенное сообщение, которое было отправлено в ${DateTime.now()}';     // Настройки сообщения     final properties = MessageProperties()..headers = headers;      // Ключ очереди     final routingKey = 'messages_queue';     // Отправляем     exchange.publish(       message,       routingKey,       properties: properties,     );     log('Отправили сообщение с задержкой $delayMs мс');   } } 

Future<Consumer> _initConsumer() async {   // Инициализируем Client   final client = Client(     settings: ConnectionSettings(       // Так как мы тестируем, используем локальный хост       host: 'localhost',       port: 5672,       authProvider: const PlainAuthenticator('guest', 'guest'),     ),   );   // Создаем клиент   final channel = await client.channel();    // Создаем очередь   final queue = await channel.queue(     'messages_queue',     durable: true,   );   // Создаем потребителя, с тегом my_consumer   final consumer = await queue.consume(     consumerTag: 'my_consumer',   );   return consumer; }

import 'dart:async';  import 'package:dart_amqp/dart_amqp.dart'; import 'package:flutter/material.dart';  void main() async {   // Инициализируем Consumer   final consumer = await _initConsumer();    runApp(MyApp(consumer: consumer)); }  Future<Consumer> _initConsumer() async {   // Инициализируем Client   final client = Client(     settings: ConnectionSettings(       // Так как мы тестируем, используем локальный хост       host: 'localhost',       port: 5672,       authProvider: const PlainAuthenticator('guest', 'guest'),     ),   );   // Создаем клиент   final channel = await client.channel();    // Создаем очередь   final queue = await channel.queue(     'messages_queue',     durable: true,   );   // Создаем потребителя, с тегом my_consumer   final consumer = await queue.consume(     consumerTag: 'my_consumer',   );   return consumer; }  class MyApp extends StatefulWidget {   const MyApp({super.key, required this.consumer});   final Consumer consumer;    @override   State<MyApp> createState() => _MyAppState(); }  class _MyAppState extends State<MyApp> {   String message = 'Нет данных';   late final StreamSubscription streamSubscription;    @override   initState() {     super.initState();     // Подписываемся на получение сообщений     streamSubscription = widget.consumer.listen((AmqpMessage amqpMessage) {       message =           'Полученное сообщение: ${amqpMessage.payloadAsString} \nВремя получения: ${DateTime.now().toIso8601String()}';        setState(() {});     });   }    @override   void dispose() {     streamSubscription.cancel();     super.dispose();   }    @override   Widget build(BuildContext context) {     return MaterialApp(       home: Scaffold(         appBar: AppBar(           title: const Text(             'Flutter Consumer Пример',           ),         ),         body: Center(child: Text(message)),       ),     );   } }