Flutter网络通信插件busrt_client的使用

本文将详细介绍如何在Flutter项目中使用busrt_client插件进行网络通信。busrt_client是一个用于与busrt服务器通信的客户端库。

插件介绍

GitHub: https://github.com/alttch/busrt
Pub.dev: https://pub.dev/packages/busrt_client

示例代码

import 'dart:isolate';
import 'dart:typed_data';

import 'package:busrt_client/busrt_client.dart';
import 'package:msgpack_dart/msgpack_dart.dart';

int myValue = 0;
int fromRpcCall = 0;
int sum = 0;

void main(List<String> arguments) async {
  final start = DateTime.now();

  // 创建两个独立的Isolate来运行不同的任务
  final f1 = Isolate.run(() => runWorker('worker.add', 'worker.minus'));
  final f2 = Isolate.run(() => runWorker('worker.minus', 'worker.add'));

  // 等待两个Isolate的任务完成
  final [res1, res2] = await Future.wait([f1, f2]);

  // 打印结果和耗时
  print([res1, res2, DateTime.now().difference(start)]);
}

// 运行一个Worker任务
Future<Map<String, int>> runWorker(String myName, String partnerName) async {
  // 初始化Bus和RPC对象
  final bus = Bus(myName);
  final rpc = Rpc(bus, onCall: onCall, onNotification: onNotification);
  await rpc.bus.connect("/tmp/busrt.sock");

  // 获取要调用的方法名称
  final method = partnerName.replaceAll('worker.', '');
  final count = method == 'add' ? 501 : 499;
  await Future.delayed(Duration(milliseconds: 10));

  // 发送一系列RPC调用
  for (var i in List.generate(count, (i) => i)) {
    final res = await rpc.call(partnerName, method, params: serialize({'value': i}));
    await res.waitCompleted();
  }

  // 发送另一个RPC调用
  await Future.delayed(Duration(milliseconds: 10));
  final res = await rpc.call(partnerName, 'get');
  fromRpcCall = deserialize((await res.waitCompleted())!.payload)['value'];

  // 发送通知
  final r = await rpc.notify(partnerName, payload: serialize({'value': myValue}));
  await r.waitCompleted();

  // 发送另一个RPC调用
  await Future.delayed(Duration(milliseconds: 10));
  final stat = await rpc.call('.broker', 'stats');
  final frame = await stat.waitCompleted();
  print(deserialize(frame!.payload));

  // 断开连接
  await rpc.bus.disconnect();

  // 返回结果
  return {"myValue": myValue, "fromRpcCall": fromRpcCall, "sum": sum};
}

// 处理通知事件
void onNotification(RpcEvent e) {
  sum = myValue + deserialize(e.payload)['value'] as int;
}

// 处理RPC调用事件
Uint8List? onCall(RpcEvent e) {
  final method = e.method;

  switch (method) {
    case 'add':
      final payload = deserialize(e.payload)['value'] as int;
      myValue += payload;
      return null;
    case 'minus':
      final payload = deserialize(e.payload)['value'] as int;
      myValue -= payload;
      return null;
    case 'get':
      return serialize({'value': myValue});
    default:
      throw RpcMethodNotFoundError(method);
  }
}

代码解释

导入必要的包：

import 'dart:isolate';
import 'dart:typed_data';
import 'package:busrt_client/busrt_client.dart';
import 'package:msgpack_dart/msgpack_dart.dart';

定义全局变量：

int myValue = 0;
int fromRpcCall = 0;
int sum = 0;

主函数：

void main(List<String> arguments) async {
  final start = DateTime.now();

  // 创建两个独立的Isolate来运行不同的任务
  final f1 = Isolate.run(() => runWorker('worker.add', 'worker.minus'));
  final f2 = Isolate.run(() => runWorker('worker.minus', 'worker.add'));

  // 等待两个Isolate的任务完成
  final [res1, res2] = await Future.wait([f1, f2]);

  // 打印结果和耗时
  print([res1, res2, DateTime.now().difference(start)]);
}

运行Worker任务：

Future<Map<String, int>> runWorker(String myName, String partnerName) async {
  // 初始化Bus和RPC对象
  final bus = Bus(myName);
  final rpc = Rpc(bus, onCall: onCall, onNotification: onNotification);
  await rpc.bus.connect("/tmp/busrt.sock");

  // 获取要调用的方法名称
  final method = partnerName.replaceAll('worker.', '');
  final count = method == 'add' ? 501 : 499;
  await Future.delayed(Duration(milliseconds: 10));

  // 发送一系列RPC调用
  for (var i in List.generate(count, (i) => i)) {
    final res = await rpc.call(partnerName, method, params: serialize({'value': i}));
    await res.waitCompleted();
  }

  // 发送另一个RPC调用
  await Future.delayed(Duration(milliseconds: 10));
  final res = await rpc.call(partnerName, 'get');
  fromRpcCall = deserialize((await res.waitCompleted())!.payload)['value'];

  // 发送通知
  final r = await rpc.notify(partnerName, payload: serialize({'value': myValue}));
  await r.waitCompleted();

  // 发送另一个RPC调用
  await Future.delayed(Duration(milliseconds: 10));
  final stat = await rpc.call('.broker', 'stats');
  final frame = await stat.waitCompleted();
  print(deserialize(frame!.payload));

  // 断开连接
  await rpc.bus.disconnect();

  // 返回结果
  return {"myValue": myValue, "fromRpcCall": fromRpcCall, "sum": sum};
}

处理通知事件：

void onNotification(RpcEvent e) {
  sum = myValue + deserialize(e.payload)['value'] as int;
}

处理RPC调用事件：

Uint8List? onCall(RpcEvent e) {
  final method = e.method;

  switch (method) {
    case 'add':
      final payload = deserialize(e.payload)['value'] as int;
      myValue += payload;
      return null;
    case 'minus':
      final payload = deserialize(e.payload)['value'] as int;
      myValue -= payload;
      return null;
    case 'get':
      return serialize({'value': myValue});
    default:
      throw RpcMethodNotFoundError(method);
  }
}

更多关于Flutter网络通信插件busrt_client的使用的实战教程也可以访问 https://www.itying.com/category-92-b0.html

zlyuanteng 1楼

更多关于Flutter网络通信插件busrt_client的使用的实战系列教程也可以访问 https://www.itying.com/category-92-b0.html

busrt_client 是一个用于 Flutter 的 Dart 插件，用于与基于 BUS/RT（一个轻量级的消息传递协议）的服务器进行通信。它允许你在 Flutter 应用中发送和接收消息，实现实时通信。

以下是使用 busrt_client 插件的基本步骤：

1. 添加依赖

首先，你需要在 pubspec.yaml 文件中添加 busrt_client 插件的依赖：

dependencies:
  flutter:
    sdk: flutter
  busrt_client: ^0.1.0  # 请使用最新版本

然后运行 flutter pub get 来获取依赖。

2. 导入插件

在你的 Dart 文件中导入 busrt_client 插件：

import 'package:busrt_client/busrt_client.dart';

3. 创建客户端并连接到服务器

你可以使用 BusrtClient 类来创建客户端并连接到 BUS/RT 服务器：

void connectToBusrtServer() async {
  // 创建客户端
  final client = BusrtClient();

  // 连接到服务器
  try {
    await client.connect('ws://your-busrt-server-address:port');
    print('Connected to BUS/RT server');
  } catch (e) {
    print('Failed to connect to BUS/RT server: $e');
  }
}

4. 发送消息

你可以使用 send 方法向服务器发送消息：

void sendMessage(BusrtClient client) async {
  try {
    await client.send('your_topic', 'Hello, BUS/RT!');
    print('Message sent');
  } catch (e) {
    print('Failed to send message: $e');
  }
}

5. 接收消息

你可以通过监听 onMessage 事件来接收来自服务器的消息：

void listenToMessages(BusrtClient client) {
  client.onMessage.listen((message) {
    print('Received message: ${message.body}');
  });
}

6. 断开连接

当你不再需要与服务器通信时，可以断开连接：

void disconnectFromServer(BusrtClient client) async {
  try {
    await client.disconnect();
    print('Disconnected from BUS/RT server');
  } catch (e) {
    print('Failed to disconnect: $e');
  }
}

7. 完整示例

以下是一个完整的示例，展示了如何连接、发送和接收消息以及断开连接：

import 'package:busrt_client/busrt_client.dart';

void main() async {
  final client = BusrtClient();

  // 连接
  try {
    await client.connect('ws://your-busrt-server-address:port');
    print('Connected to BUS/RT server');
  } catch (e) {
    print('Failed to connect to BUS/RT server: $e');
    return;
  }

  // 监听消息
  client.onMessage.listen((message) {
    print('Received message: ${message.body}');
  });

  // 发送消息
  try {
    await client.send('your_topic', 'Hello, BUS/RT!');
    print('Message sent');
  } catch (e) {
    print('Failed to send message: $e');
  }

  // 断开连接
  try {
    await client.disconnect();
    print('Disconnected from BUS/RT server');
  } catch (e) {
    print('Failed to disconnect: $e');
  }
}